Gabriel39 commented on code in PR #65135:
URL: https://github.com/apache/doris/pull/65135#discussion_r3557772840
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/source/IcebergScanNode.java:
##########
@@ -896,14 +967,115 @@ private Split createIcebergSplit(FileScanTask
fileScanTask) {
return split;
}
- private Split createIcebergSysSplit(FileScanTask fileScanTask) {
- long rowCount = fileScanTask.file() == null ? 1 :
fileScanTask.file().recordCount();
+ private Split createIcebergSysSplit(ScanTask scanTask) {
+ long rowCount = Math.max(scanTask.estimatedRowsCount(), 1L);
+ if (scanTask.isFileScanTask() && scanTask.asFileScanTask().file() !=
null) {
+ rowCount =
Math.max(scanTask.asFileScanTask().file().recordCount(), 1L);
+ }
IcebergSplit split = IcebergSplit.newSysTableSplit(
- SerializationUtil.serializeToBase64(fileScanTask), rowCount);
+ SerializationUtil.serializeToBase64(scanTask), rowCount);
split.setTableFormatType(TableFormatType.ICEBERG);
return split;
}
+ private Split createIcebergPositionDeleteSysSplit(PositionDeletesScanTask
task) {
+ DeleteFile deleteFile = task.file();
+ String originalPath = deleteFile.path().toString();
+ LocationPath locationPath = createLocationPathWithCache(originalPath);
+ IcebergSplit split = IcebergSplit.newPositionDeleteSysTableSplit(
+ locationPath, task.start(), task.length(),
deleteFile.fileSizeInBytes(),
+ storagePropertiesMap, originalPath);
+ split.setTableFormatType(TableFormatType.ICEBERG);
+
split.setPositionDeleteFileFormat(getNativePositionDeleteFileFormat(deleteFile.format()));
+ split.setPositionDeleteOriginalPath(originalPath);
+ if (deleteFile.format() == FileFormat.PUFFIN) {
+
split.setPositionDeleteContent(IcebergDeleteFileFilter.DeletionVector.type());
+
split.setPositionDeleteReferencedDataFilePath(deleteFile.referencedDataFile());
+ split.setPositionDeleteContentOffset(deleteFile.contentOffset());
+
split.setPositionDeleteContentSizeInBytes(deleteFile.contentSizeInBytes());
+ } else {
+
split.setPositionDeleteContent(IcebergDeleteFileFilter.PositionDelete.type());
+ }
+
+ split.setPartitionSpecId(deleteFile.specId());
+ PartitionSpec partitionSpec =
icebergTable.specs().get(deleteFile.specId());
+ if (partitionSpec != null && partitionSpec.isPartitioned() &&
deleteFile.partition() != null) {
+ split.setPartitionDataJson(getPartitionDataObjectJson(
+ (PartitionData) deleteFile.partition(), partitionSpec,
+ getPositionDeletesOutputPartitionFields()));
+ }
+ return split;
+ }
+
+ @SuppressWarnings("unchecked")
+ private Iterable<PositionDeletesScanTask>
splitPositionDeleteScanTask(PositionDeletesScanTask task) {
+ return ((SplittableScanTask<PositionDeletesScanTask>)
task).split(targetSplitSize);
+ }
+
+ private TFileFormatType getNativePositionDeleteFileFormat(FileFormat
fileFormat) {
+ if (fileFormat == FileFormat.PARQUET || fileFormat ==
FileFormat.PUFFIN) {
+ return TFileFormatType.FORMAT_PARQUET;
+ } else if (fileFormat == FileFormat.ORC) {
+ return TFileFormatType.FORMAT_ORC;
+ }
+ throw new UnsupportedOperationException(
+ "Unsupported Iceberg position delete file format: " +
fileFormat);
+ }
+
+ private List<NestedField> getPositionDeletesOutputPartitionFields() {
+ NestedField partitionField =
icebergTable.schema().findField("partition");
+ Preconditions.checkNotNull(partitionField,
+ "Partition field not found in Iceberg position_deletes
metadata table schema");
+ return partitionField.type().asNestedType().fields();
+ }
+
+ private String getPartitionDataObjectJson(PartitionData partitionData,
PartitionSpec partitionSpec,
+ List<NestedField> outputPartitionFields) {
+ List<String> partitionValues = IcebergUtils.getPartitionValues(
+ partitionData, partitionSpec, sessionVariable.getTimeZone());
+ List<NestedField> partitionTypes =
partitionData.getPartitionType().asNestedType().fields();
+ Map<String, Object> partitionValueByName = new HashMap<>();
+ List<PartitionField> fields = partitionSpec.fields();
+ for (int i = 0; i < fields.size(); i++) {
+ partitionValueByName.put(fields.get(i).name(),
Review Comment:
[P1] Please key this mapping by partition field ID rather than field name.
Iceberg keeps the field ID when a partition field is renamed, while the unified
`partition` struct uses the most recent name. After writing a delete file under
`p1` and then running `RENAME PARTITION FIELD p1 TO p2`, this old spec
contributes `p1` here but the output loop looks up `p2`, so the value is
silently serialized as NULL. The current ADD-partition-field test does not
cover this. Please map the transformed spec fields to the output fields by ID
and add a delete-before/after-partition-rename regression.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/source/IcebergScanNode.java:
##########
@@ -896,14 +967,115 @@ private Split createIcebergSplit(FileScanTask
fileScanTask) {
return split;
}
- private Split createIcebergSysSplit(FileScanTask fileScanTask) {
- long rowCount = fileScanTask.file() == null ? 1 :
fileScanTask.file().recordCount();
+ private Split createIcebergSysSplit(ScanTask scanTask) {
+ long rowCount = Math.max(scanTask.estimatedRowsCount(), 1L);
+ if (scanTask.isFileScanTask() && scanTask.asFileScanTask().file() !=
null) {
+ rowCount =
Math.max(scanTask.asFileScanTask().file().recordCount(), 1L);
+ }
IcebergSplit split = IcebergSplit.newSysTableSplit(
- SerializationUtil.serializeToBase64(fileScanTask), rowCount);
+ SerializationUtil.serializeToBase64(scanTask), rowCount);
split.setTableFormatType(TableFormatType.ICEBERG);
return split;
}
+ private Split createIcebergPositionDeleteSysSplit(PositionDeletesScanTask
task) {
+ DeleteFile deleteFile = task.file();
+ String originalPath = deleteFile.path().toString();
+ LocationPath locationPath = createLocationPathWithCache(originalPath);
+ IcebergSplit split = IcebergSplit.newPositionDeleteSysTableSplit(
+ locationPath, task.start(), task.length(),
deleteFile.fileSizeInBytes(),
+ storagePropertiesMap, originalPath);
+ split.setTableFormatType(TableFormatType.ICEBERG);
+
split.setPositionDeleteFileFormat(getNativePositionDeleteFileFormat(deleteFile.format()));
+ split.setPositionDeleteOriginalPath(originalPath);
+ if (deleteFile.format() == FileFormat.PUFFIN) {
+
split.setPositionDeleteContent(IcebergDeleteFileFilter.DeletionVector.type());
+
split.setPositionDeleteReferencedDataFilePath(deleteFile.referencedDataFile());
+ split.setPositionDeleteContentOffset(deleteFile.contentOffset());
+
split.setPositionDeleteContentSizeInBytes(deleteFile.contentSizeInBytes());
+ } else {
+
split.setPositionDeleteContent(IcebergDeleteFileFilter.PositionDelete.type());
+ }
+
+ split.setPartitionSpecId(deleteFile.specId());
+ PartitionSpec partitionSpec =
icebergTable.specs().get(deleteFile.specId());
+ if (partitionSpec != null && partitionSpec.isPartitioned() &&
deleteFile.partition() != null) {
+ split.setPartitionDataJson(getPartitionDataObjectJson(
+ (PartitionData) deleteFile.partition(), partitionSpec,
+ getPositionDeletesOutputPartitionFields()));
+ }
+ return split;
+ }
+
+ @SuppressWarnings("unchecked")
+ private Iterable<PositionDeletesScanTask>
splitPositionDeleteScanTask(PositionDeletesScanTask task) {
+ return ((SplittableScanTask<PositionDeletesScanTask>)
task).split(targetSplitSize);
+ }
+
+ private TFileFormatType getNativePositionDeleteFileFormat(FileFormat
fileFormat) {
+ if (fileFormat == FileFormat.PARQUET || fileFormat ==
FileFormat.PUFFIN) {
Review Comment:
[P2] This rejects valid Avro position-delete files. Iceberg supports
`FileFormat.AVRO` for position deletes (including files with and without the
optional `row` payload), but this method throws before any split is created. As
a result, `$position_deletes` cannot be queried for a table using an Avro
delete-file format. Please either add an Avro reader path or fall back to the
serialized JNI scan for these tasks, and add an Avro position-delete regression.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
