github-actions[bot] commented on code in PR #67687:
URL: https://github.com/apache/doris/pull/67687#discussion_r3964216049
##########
fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergScanPlanProviderTest.java:
##########
@@ -1832,6 +1859,52 @@ public void
planScanPinnedToOlderSnapshotReadsOnlyThatSnapshotsFiles() {
Assertions.assertTrue(pinned.get(0).getPath().get().endsWith("f1.parquet"));
}
+ @Test
+ public void planScanHistoricalPredicateSurvivesColumnRename() {
+ assertHistoricalPredicatePlansAfterSchemaEvolution(false);
+ }
+
+ @Test
+ public void planScanHistoricalPredicateSurvivesColumnDrop() {
+ assertHistoricalPredicatePlansAfterSchemaEvolution(true);
+ }
+
+ private void assertHistoricalPredicatePlansAfterSchemaEvolution(boolean
dropColumn) {
+ Schema historicalSchema = new Schema(
+ Types.NestedField.optional(1, "x", Types.IntegerType.get()),
+ Types.NestedField.optional(2, "y", Types.IntegerType.get()),
+ Types.NestedField.optional(3, "part",
Types.IntegerType.get()));
+ Table table = createTable(
+ "historical_predicate_after_" + (dropColumn ? "drop" :
"rename"),
+ historicalSchema, PartitionSpec.unpartitioned(),
+ Collections.singletonMap(TableProperties.FORMAT_VERSION, "2"));
+ table.newFastAppend()
+ .appendFile(dataFile(table.spec(),
"s3://b/db/historical.parquet", 1024, null, null))
+ .commit();
+ long historicalSnapshotId = table.currentSnapshot().snapshotId();
+ int historicalSchemaId = table.currentSnapshot().schemaId();
+
+ if (dropColumn) {
+ table.updateSchema().deleteColumn("x").commit();
+ } else {
+ table.updateSchema().renameColumn("x", "renamed_x").commit();
+ }
+ table.newFastAppend()
Review Comment:
[P1] Cover the schema-only state before advancing the snapshot. A
rename/drop metadata commit does not create a new Iceberg snapshot, so at this
point historicalSnapshotId is still table.currentSnapshot().snapshotId().
Iceberg 1.11's SnapshotScan.specs() then skips historical rebinding and returns
specs bound to the renamed/dropped current schema, causing the old-name x
predicate to fail during projection. This append makes the IDs differ and lets
both tests avoid that unresolved production case. Please plan and assert
immediately before the append, then fix the equal-snapshot/schema-evolved path.
##########
fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergPartitionUtils.java:
##########
@@ -805,11 +803,9 @@ private static IcebergRawPartition
generateRawPartition(Table table, StructLike
for (int i = 0; i < partitionSpec.fields().size(); ++i) {
PartitionField partitionField = partitionSpec.fields().get(i);
Class<?> fieldClass = partitionSpec.javaClasses()[i];
- int fieldId = partitionField.fieldId();
- // Iceberg partition field id starts at PARTITION_DATA_ID_START,
so the index into partitionData is
- // fieldId - PARTITION_DATA_ID_START.
- int index = fieldId - PARTITION_DATA_ID_START;
- Object o = partitionData.get(index, fieldClass);
+ // A spec's partition struct is compact even when evolved field
IDs have gaps, so index by the
+ // field's position in this spec rather than by its table-global
partition field ID.
+ Object o = partitionData.get(i, fieldClass);
Review Comment:
[P2] Index this table-wide partition struct by partition field ID. Iceberg's
PartitionsTable first coerces every file partition into
Partitioning.partitionType(table), the union across all specs; this row is not
compact for its own spec. In the new test, that struct is [region_bucket, id],
so the current spec's sole id field is still slot 1 and get(0) returns the
missing region value (null), rendering id=null. Build a
field-ID-to-unified-ordinal map and assert the evolved partitions'
names/values, not just their count.
##########
fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/iceberg/DeleteFileIndex.java:
##########
@@ -406,6 +410,11 @@ Builder afterSequenceNumber(long seq) {
return this;
}
+ Builder schemasById(Map<Integer, Schema> newSchemasById) {
Review Comment:
[P1] Pass the full schema history into this builder. Iceberg 1.11's
DataTableScan always supplies schemas() before building its delete index, but
this method is still package-private and Doris's manifest-cache path can only
pass specsById. After reloading a table where an equality-delete key was
dropped, the persisted specs are bound to the current schema, so the fallback
map cannot resolve that field and lazy streaming planning throws instead of
reaching the SDK fallback. Make this method public, call
schemasById(table.schemas()) from cacheBackedFileScanTasks, and cover a
cache-on persisted equality-delete evolution case.
##########
fe/pom.xml:
##########
@@ -360,7 +360,7 @@ under the License.
<!-- ATTN: avro version must be consistent with Iceberg version -->
<!-- Please modify iceberg.version and avro.version together,
you can find avro version info in iceberg mvn repository -->
- <iceberg.version>1.10.1</iceberg.version>
+ <iceberg.version>1.11.0</iceberg.version>
Review Comment:
[P1] Preserve the system-table task wire format across this upgrade. FE
Java-serializes Iceberg FileScanTask objects, including Schema, while the
selected BE deserializes them using its locally packaged Iceberg classes.
Schema has no explicit serialVersionUID, and 1.10.1/1.11.0 task graphs fail
deserialization in both directions, so generic $system_table queries break
whenever a rolling cluster pairs a new FE with an old BE or vice versa; only
$position_deletes avoids this JNI path. Please use a stable/versioned
representation or add mixed-version routing support and coverage before
changing both artifacts.
##########
regression-test/suites/external_table_p0/iceberg/iceberg_schema_change_ddl_with_branch.groovy:
##########
@@ -198,10 +198,11 @@ suite("iceberg_schema_change_ddl_with_branch",
"p0,external") {
// All branches expose the current table columns: id, name, grade, email,
phone.
- // Verify all branches have the latest columns
- qt_all_branches_have_grade """ SELECT id, grade FROM
${branch_table_name}@branch(branch1) WHERE grade > 0 ORDER BY id """
+ // Iceberg validates filters against the referenced snapshot schema, so
columns renamed or
+ // added later are verified through projection instead of predicates on
historical branches.
+ qt_all_branches_have_grade """ SELECT id, grade FROM
${branch_table_name}@branch(branch1) ORDER BY id """
Review Comment:
[P2] Keep predicate coverage for branch-current-schema columns. Iceberg
1.11's SnapshotUtil.schemaFor(table, ref) deliberately returns table.schema()
for branches, and this master connector preserves the ref through
ConnectorMvccSnapshot and calls scan.useRef. These grade/phone predicates are
therefore valid here; replacing them with projection imports a source-branch
workaround that master does not need and removes the only oracle for filter
planning on renamed/later-added branch columns. Please restore the predicates
(or equivalent cases).
--
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]
