Gabriel39 commented on code in PR #67904: URL: https://github.com/apache/doris/pull/67904#discussion_r4011696716
########## 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: Fixed in 31945b11ac. This is within the statement-generation guarantee introduced by this PR. The immutable statement pin now captures the fallback child's selected snapshot ID and content digest, using the SDK's timestamp/FIRST_SNAPSHOT_ID selection once at pin creation. Later table copies apply that captured child ID instead of translating the main timestamp again. An absent selected snapshot is recorded and a subsequent append is rejected. Tests cover same-schema branch recreation with different rows, no eligible timestamp snapshot, and first append after an absent fallback snapshot. ########## 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: The reported regression is valid, but adding another Iceberg coordinate would keep expanding a Paimon fix into a general metadata-publication redesign. 31945b11ac removes the cause instead: latest-schema publication now requires an explicit typed opt-in, which only Paimon enables. A positive schema ID alone leaves ordinary schema/DDL publication unchanged. All collateral Iceberg and Hive edits have been removed; those directories match the merged upstream base. The generic FE regression test verifies that a non-opted-in pin neither publishes a retained schema nor calls the at-snapshot schema loader. Existing historical time-travel metadata policies remain outside this PR. ########## 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: Fixed in 31945b11ac. The reader-option checks now use five unique order_qt_ labels for warmup, Native catalog options, Native relation overrides, JNI catalog options, and JNI relation overrides. The .out entries were generated by the regression framework and the complete suite was rerun against the final FE build. -- 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]
