github-actions[bot] commented on code in PR #68902:
URL: https://github.com/apache/doris/pull/68902#discussion_r4240685155
##########
fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergProcedureOps.java:
##########
@@ -249,6 +254,35 @@ private List<ConnectorRewriteGroup>
planInAuthScopeUntracked(IcebergTableHandle
return
groups.stream().map(IcebergProcedureOps::toConnectorRewriteGroup).collect(Collectors.toList());
}
+ /**
+ * Routes the procedure table's {@code io()} through the plugin-side
Kerberos {@code doAs} — the procedure-path
+ * mirror of {@code IcebergScanPlanProvider.wrapTableForScan} and
+ * {@code IcebergConnectorTransaction.openTransaction}. The {@code
executeAuthenticated} scope around the body
+ * covers only the calling thread, while iceberg reads and writes metadata
files on its shared worker pool:
+ * {@code expire_snapshots} reads the expired snapshots' manifest lists
there after committing (CI:
+ * test_iceberg_hadoop_catalog_kerberos failed "Failed to open input
stream for file: .../snap-*.avro"),
+ * {@code cherrypick_snapshot} writes the picked files' manifest there,
and rewrite planning reads the manifests
+ * of a multi-manifest table there. The wrapped FileIO keeps prefix
listing for {@code remove_orphan_files}
+ * ({@link IcebergAuthenticatedFileIO#withPrefixOperations}).
+ *
+ * <p>Non-Kerberos catalogs ({@code getPluginAuthenticator() == null}) and
offline tests (plain context /
+ * non-{@link BaseTable} fakes) get the table unchanged. Package-private
for unit testing.
+ */
+ Table wrapTableForProcedure(Table table) {
+ if (!(context instanceof TcclPinningConnectorContext) || !(table
instanceof BaseTable)) {
+ return table;
+ }
+ HadoopAuthenticator auth = ((TcclPinningConnectorContext)
context).getPluginAuthenticator();
+ if (auth == null) {
+ return table;
+ }
+ BaseTable baseTable = (BaseTable) table;
+ TableOperations rawOps = baseTable.operations();
+ return new BaseTable(new IcebergAuthenticatedTableOperations(rawOps,
Review Comment:
[P2] Preserve REST planning for zero-group rewrites. A Kerberos REST catalog
using server scan planning loads a `RESTTable`, whose `newScan()` plans through
the REST endpoint. Before this wrapper, `rewrite_data_files` could return its
all-zero result when no files qualify, before the downstream scan that rejects
server planning. Replacing the table with a plain `BaseTable` makes even that
no-op plan read manifests locally; it can fail when access is available only
through the REST plan response's credentials. Keep the specialized scan
behavior for planning, or explicitly reject this mode before local manifest I/O.
##########
regression-test/suites/external_table_p0/kerberos/test_iceberg_hadoop_catalog_kerberos.groovy:
##########
@@ -97,6 +97,72 @@ suite("test_iceberg_hadoop_catalog_kerberos", "p0,external")
{
'partition_val2'
);
"""
+ def snapshotsAfterFirstInsert = sql """
+ select snapshot_id from ${test_tbl_name}\$snapshots order by
committed_at
+ """
+ assert snapshotsAfterFirstInsert.size() == 1
+ String firstSnapshotId = String.valueOf(snapshotsAfterFirstInsert[0][0])
+
+ sql """
+ insert into ${test_tbl_name} values (
+ '2024-05-27 12:34:56',
+ false,
+ 456,
+ 4567890123456,
+ 45.67,
+ 89.012,
+ 45678.9012,
+ 'another example',
+ '2024-05-27',
+ '2024-05-27 14:00:00',
+ 'partition_val1',
+ 'partition_val2'
+ );
+ """
+ def dataBeforeRollback = sql """select count(1) from ${test_tbl_name}"""
+ assert dataBeforeRollback.get(0).get(0) == 2
+
+ // Rollback must commit metadata with the same Kerberos identity used to
acquire the table.
+ sql """
+ alter table ${catalog_name}.${database_name}.${test_tbl_name}
+ execute rollback_to_snapshot("snapshot_id" = "${firstSnapshotId}")
+ """
+ def dataAfterRollback = sql """select count(1) from ${test_tbl_name}"""
+ assert dataAfterRollback.get(0).get(0) == 1
+
+ sql """
+ insert into ${test_tbl_name} values (
+ '2024-05-27 12:34:56',
+ false,
+ 456,
+ 4567890123456,
+ 45.67,
+ 89.012,
+ 45678.9012,
+ 'another example',
+ '2024-05-27',
+ '2024-05-27 14:00:00',
+ 'partition_val1',
+ 'partition_val2'
+ );
+ """
+
+ def snapshotsBeforeExpire = sql """
+ select snapshot_id from ${test_tbl_name}\$snapshots order by
committed_at
+ """
+ assert snapshotsBeforeExpire.size() == 3
+
+ // The action must reuse the catalog's Kerberos identity while committing
new metadata.
+ sql """
+ alter table ${catalog_name}.${database_name}.${test_tbl_name}
+ execute expire_snapshots("retain_last" = "1")
+ """
+
+ def snapshotsAfterExpire = sql """
+ select snapshot_id from ${test_tbl_name}\$snapshots order by
committed_at
+ """
+ assert snapshotsAfterExpire.size() == 1
+
def dataResult = sql """select count(1) from ${test_tbl_name} """
- assert dataResult.get(0).get(0) == 1
+ assert dataResult.get(0).get(0) == 2
Review Comment:
[P2] Verify surviving rows with a recorded data query. `COUNT(1)` is
eligible for Iceberg's metadata count pushdown here, so this assertion can
return the manifest count of 2 without opening either retained ORC file.
Cleanup could delete a live file, or preserve two wrong rows, while the
snapshot and count assertions remain green. Add an ordered `qt_sql` over
identifying columns after expiration with a runner-generated result, and record
the other fixed expected counts through the same mechanism as required by
`AGENTS.md`.
--
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]
