github-actions[bot] commented on code in PR #67904: URL: https://github.com/apache/doris/pull/67904#discussion_r4011430744
########## fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonSchemaPin.java: ########## @@ -0,0 +1,146 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package org.apache.doris.connector.paimon; + +import org.apache.doris.connector.spi.DorisConnectorException; + +import com.google.common.hash.Hashing; +import org.apache.paimon.schema.TableSchema; +import org.apache.paimon.table.DataTable; +import org.apache.paimon.table.DelegatedFileStoreTable; +import org.apache.paimon.table.FallbackReadFileStoreTable; +import org.apache.paimon.table.Table; +import org.apache.paimon.utils.SnapshotManager; + +import java.nio.charset.StandardCharsets; +import java.util.HashMap; +import java.util.Map; +import java.util.function.LongSupplier; + +/** Immutable file coordinates carried by the MVCC pin, including across INSERT planning attempts. */ +final class PaimonSchemaPin { + private static final String PREFIX = "doris.internal.paimon.schema-pin."; + + private PaimonSchemaPin() {} + + static Map<String, String> capture(Table table, long schemaId, long snapshotId) { + Map<String, String> pin = new HashMap<>(); + if (schemaId >= 0) { + capture(table, schemaId, snapshotId, "", pin); + if (!pin.isEmpty()) { + pin.put(PREFIX + "shape", shape(table)); + } + } + return pin; + } + + private static void capture(Table table, long schemaId, long snapshotId, String path, Map<String, String> pin) { + if (table instanceof FallbackReadFileStoreTable) { + FallbackReadFileStoreTable pair = (FallbackReadFileStoreTable) table; + capture(pair.wrapped(), schemaId, snapshotId, path, pin); + TableSchema fallback = pair.fallback().schemaManager().latest().orElseThrow(IllegalStateException::new); + capture(pair.fallback(), fallback.id(), -1, path + "fallback.", pin); Review Comment: [P1] Fence the fallback child's data generation here as well. Passing `-1` means its pin contains only a schema digest, so deleting/recreating that branch with the same schema can pass `validate()` even though its data history is a different generation. [Paimon 1.3.1's fallback rewrite](https://github.com/apache/paimon/blob/release-1.3.1/paimon-core/src/main/java/org/apache/paimon/table/FallbackReadFileStoreTable.java#L149-L174) then translates the main `scan.snapshot-id` by timestamp and, when the replacement has no snapshot old enough, selects that child's `FIRST_SNAPSHOT_ID`; the already-pinned statement can consequently read replacement fallback rows instead of failing. This is separate from the earlier fallback-schema thread and the main-child generation fence. Please retain and validate a fallback-local snapshot/content coordinate (including the no-eligible-snapshot state), and cover same-schema fallback recreation with different rows. ########## fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergConnectorMetadata.java: ########## @@ -483,33 +488,65 @@ public ConnectorTableSchema getTableSchema( return getTableSchema(session, handle); } Table table = loadTable(session, iceHandle); - Schema schema; - if (table.currentSnapshot() == null) { - // Empty table: legacy getSchema falls back to the latest schema (NEWEST_SCHEMA_ID path). - schema = table.schema(); - } else { - schema = table.schemas().get((int) snapshot.getSchemaId()); - if (schema == null) { - // Defensive: a pinned id absent from table.schemas() (legacy would NPE) -> latest. - // INVARIANT: this SLOT-schema fallback MUST stay identical to the DICT-schema fallback in - // IcebergScanPlanProvider.pinnedSchema (same getSchemaId() lookup + same silent -> table.schema()). - // If the two diverge, the field-id dict names and the BE scan-slot names resolve DIFFERENT - // schemas -> BE children.at() std::out_of_range-SIGABRT on a schema-evolved time-travel read - // (reverify #65185 L16). Do not harden ONE side to throw without the other. - schema = table.schema(); + validateSnapshotTable(iceHandle, table, snapshot); + Schema schema = resolvePinnedSchema(table, snapshot); + String specId = snapshot.getProperties().get(PARTITION_SPEC_ID_PROPERTY); + PartitionSpec spec = specId == null ? table.spec() : table.specs().get(Integer.parseInt(specId)); + if (spec == null) { + // Keep the legacy missing-history fallback after checking the table identity. + spec = table.spec(); + } + return buildTableSchema(iceHandle.getTableName(), table, schema, spec, true); Review Comment: [P1] Pin the sort order together with this retained schema/spec coordinate. This call now assembles old pinned columns and partition metadata with a live `Table`; `buildShowSortClause(table)` still reads that table's current sort order and schema. In the REST-vended/positive-TTL case, `/api/query_schema` analyzes the query (installing the cached S1 pin) and then calls `Env.getDdlStmt` while that `StatementContext` is still active, so a later schema/sort-order commit can produce DDL with S1 columns but `ORDER BY` an M2-only column. This is distinct from the earlier partition/spec thread. Please carry/resolve the sort-order ID with the latest pin and cover the query-schema DDL path after schema plus sort-order evolution. ########## regression-test/suites/external_table_p0/paimon/test_paimon_schema_only_snapshot_precision.groovy: ########## @@ -0,0 +1,147 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +suite("test_paimon_schema_only_snapshot_precision", "p0,external,paimon") { + String enabled = context.config.otherConfigs.get("enablePaimonTest") + if (enabled == null || !enabled.equalsIgnoreCase("true")) { + logger.info("disable paimon test") + return + } + + String minioPort = context.config.otherConfigs.get("iceberg_minio_port") + String externalEnvIp = context.config.otherConfigs.get("externalEnvIp") + String catalogName = "test_paimon_schema_only_snapshot_precision" + String dbName = "paimon_schema_only_snapshot_precision_db" + String tableName = "schema_only_timeline" + String branchName = "schema_only_branch" + + def latestSnapshotId = { + List<List<Object>> rows = spark_paimon """ + select snapshot_id + from paimon.${dbName}.`${tableName}\$snapshots` + order by snapshot_id desc + limit 1 + """ + assertEquals(1, rows.size()) + return rows[0][0].toString() + } + + sql """drop catalog if exists ${catalogName}""" + sql """ + create catalog ${catalogName} properties ( + 'type'='paimon', + 'warehouse'='s3://warehouse/wh', + 's3.endpoint'='http://${externalEnvIp}:${minioPort}', + 's3.access_key'='admin', + 's3.secret_key'='password', + 's3.path.style.access'='true', + 'paimon.table-option.read.batch-size'='64', + 'meta.cache.paimon.table.ttl-second'='0' + ) + """ + + try { + // Explicit NTZ makes this exercise predicate pushdown instead of LTZ residual filtering. + spark_paimon_multi """ + create database if not exists paimon.${dbName}; + drop table if exists paimon.${dbName}.${tableName}; + create table paimon.${dbName}.${tableName} ( + id int, + old_name string, + event_time timestamp_ntz + ) using paimon + tblproperties ('file.format'='parquet'); + insert into paimon.${dbName}.${tableName} + values (1, 'base', timestamp_ntz '2024-01-01 00:00:00.123456'); + """ + String dataSnapshotId = latestSnapshotId() + spark_paimon_multi """ + call paimon.sys.create_tag( + table => '${dbName}.${tableName}', + tag => 'schema_base' + ); + call paimon.sys.create_branch( + '${dbName}.${tableName}', + '${branchName}', + 'schema_base' + ); + alter table paimon.${dbName}.`${tableName}\$branch_${branchName}` + rename column old_name to branch_name; + alter table paimon.${dbName}.${tableName} + rename column old_name to current_name; + """ + + // A schema-only rename must leave the data snapshot unchanged; otherwise these queries + // would not exercise the split between current schema binding and snapshot-pinned data. + assertEquals(dataSnapshotId, latestSnapshotId()) + + sql """switch ${catalogName}""" + sql """use ${dbName}""" + sql """refresh table ${tableName}""" + + order_qt_plain_schema """ + select id, current_name from ${tableName} order by id + """ + order_qt_options_schema """ + select id, current_name + from ${tableName}@options('scan.plan-sort-partition'='true') + order by id + """ + order_qt_branch_schema """ + select id, branch_name + from ${tableName}@branch(${branchName}) + order by id + """ + + // Sub-millisecond precision must survive FE predicate conversion or Paimon's file + // statistics can reject the only matching file before either reader sees it. + sql """set force_jni_scanner=false""" + order_qt_native_precision """ + select id from ${tableName} + where event_time = cast('2024-01-01 00:00:00.123456' as datetime(6)) + """ + sql """set force_jni_scanner=true""" + order_qt_jni_precision """ + select id from ${tableName} + where event_time = cast('2024-01-01 00:00:00.123456' as datetime(6)) + """ + + // Catalog reader policy must survive schema restoration even when it matched the old + // physical value. Relation overrides still take precedence for both reader paths. + String readerTable = "${tableName}_reader_options" + spark_paimon_multi """ + drop table if exists paimon.${dbName}.${readerTable}; + create table paimon.${dbName}.${readerTable} (id int) using paimon + tblproperties ('file.format'='parquet', 'read.batch-size'='64'); + insert into paimon.${dbName}.${readerTable} values (1); + """ + assertEquals([[1]], sql("select id from ${readerTable}")) Review Comment: [P2] Record this newly added reader-option block through the regression harness. This is distinct from the earlier thread: that thread's five plain/OPTIONS/branch/native/JNI checks were converted at lines 96-121 and now have generated `.out` entries, whereas the later checks here and at lines 138-141 cover the new catalog/relation option-precedence behavior but again use direct `assertEquals(sql(...))`. Root `AGENTS.md` requires determined results to use `qt_`/`order_qt_`; please give the native/JNI and override cases unique `order_qt_` labels and regenerate the output file. -- 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]
