[VL] Make the filename written by Iceberg Native consistent with in Java

[Zouxxyy]

What changes are proposed in this pull request?

Follow the format of filename written by iceberg java

// {partitionId:05d}-{taskId}-{operationId}-{fileCount:05d}{suffix}
00000-0-00e4a92d-4b72-4950-b562-818bfe0853e2-0-00001.parquet

How was this patch tested?

[github-actions]

Run Gluten Clickhouse CI on x86

[github-actions]

Run Gluten Clickhouse CI on x86

[github-actions]

Run Gluten Clickhouse CI on x86

[github-actions]

Run Gluten Clickhouse CI on x86

[Zouxxyy]

CC @jinchengchenghh for a look, thanks

[jinchengchenghh]

Thanks for your enhancement!

[jinchengchenghh]

The CH CI failure is not related to your PR, let us wait for the community feedback, I have raised this issue.

[jinchengchenghh]

Run Gluten Clickhouse CI on x86

[jinchengchenghh]

Please rebase to fix the CH CI

[github-actions]

Run Gluten Clickhouse CI on x86

[jinchengchenghh]

https://github.com/apache/incubator-gluten/blob/main/cpp/velox/compute/VeloxRuntime.h#L45C25-L45C38 we can get these two variables in native side by VeloxRuntime

[Zouxxyy]

I tried obtain them here

std::shared_ptr<IcebergWriter> VeloxRuntime::createIcebergWriter(
    RowTypePtr rowType,
    int32_t format,
    const std::string& outputDirectory,
    facebook::velox::common::CompressionKind compressionKind,
    const std::string& operationId,
    std::shared_ptr<const facebook::velox::connector::hive::iceberg::IcebergPartitionSpec> spec,
    const gluten::IcebergNestedField& protoField,
    const std::unordered_map<std::string, std::string>& sparkConfs) {
  GLUTEN_CHECK(taskInfo_.has_value(), "Task info must be set before creating IcebergWriter");
  auto veloxPool = memoryManager()->getLeafMemoryPool();
  auto connectorPool = memoryManager()->getAggregateMemoryPool();
  return std::make_shared<IcebergWriter>(
      rowType, format, outputDirectory, compressionKind, taskInfo_->partitionId, taskInfo_->taskId, operationId, spec, protoField, sparkConfs, veloxPool, connectorPool);
}

but got this error, perhaps the timing of setting taskInfo is inconsistent with Iceberg's write creation.

22:18:47.228 ERROR org.apache.spark.task.TaskResources: Task 0 failed by error: 
org.apache.gluten.exception.GlutenException: Task info must be set before creating IcebergWriter
	at org.apache.gluten.execution.IcebergWriteJniWrapper.init(Native Method)
	at org.apache.gluten.connector.write.IcebergDataWriteFactory.getJniWrapper(IcebergDataWriteFactory.scala:103)
	at org.apache.gluten.connector.write.IcebergDataWriteFactory.createWriter(IcebergDataWriteFactory.scala:77)
	at org.apache.spark.sql.datasources.v2.WritingColumnarBatchSparkTask.run(ColumnarWriteToDataSourceV2Exec.scala:49)
	at org.apache.spark.sql.datasources.v2.WritingColumnarBatchSparkTask.run$(ColumnarWriteToDataSourceV2Exec.scala:39)
	at org.apache.spark.sql.datasources.v2.DataWritingColumnarBatchSparkTask$.run(ColumnarWriteToDataSourceV2Exec.scala:93)

Actually, I think maybe my implementation is fine too, especially the adjustment to the ColumnarBatchDataWriterFactory interface—it now fully aligns with Spark's DataWriterFactory.

Gluten's ColumnarBatchDataWriterFactory

public interface ColumnarBatchDataWriterFactory extends Serializable {
  DataWriter<ColumnarBatch> createWriter(int partitionId, long taskId);
}

Spark's DataWriterFactory

public interface DataWriterFactory extends Serializable {
  DataWriter<InternalRow> createWriter(int partitionId, long taskId);
}

[Zouxxyy]

@jinchengchenghh Can you have a look again, thanks

[jinchengchenghh]

Looks good!