Gabriel39 commented on code in PR #66345:
URL: https://github.com/apache/doris/pull/66345#discussion_r3717619328
##########
fe/fe-connector/fe-connector-hive/src/main/java/org/apache/doris/connector/hive/HiveWritePlanProvider.java:
##########
@@ -99,19 +100,75 @@ public ConnectorSinkPlan planWrite(ConnectorSession
session, ConnectorWriteHandl
HiveTableHandle tableHandle = (HiveTableHandle)
handle.getTableHandle();
HiveConnectorTransaction transaction = currentTransaction(session);
- // Load the table under the catalog auth context; it drives both the
location resolution
- // (buildWriteContext) and the sink assembly (buildSink). beginWrite
re-loads it for its own
- // begin-guard — the double-load is accepted (mirrors iceberg),
keeping the flow simple.
+ // One fresh table generation drives validation, transaction state,
location resolution, and sink
+ // assembly. Reloading in beginWrite would reopen a TOCTOU window
after the schema fence.
HmsTableInfo table = loadTable(tableHandle);
+ validateBoundWriteMetadata(table, handle);
HiveWriteContext writeContext = buildWriteContext(session,
tableHandle, table, handle);
- transaction.beginWrite(session, tableHandle.getDbName(),
tableHandle.getTableName(), writeContext);
+ transaction.beginWrite(session, tableHandle.getDbName(),
tableHandle.getTableName(), writeContext, table);
THiveTableSink sink = buildSink(session, tableHandle, table, handle,
writeContext);
TDataSink dataSink = new TDataSink(TDataSinkType.HIVE_TABLE_SINK);
dataSink.setHiveTableSink(sink);
return new ConnectorSinkPlan(dataSink);
}
+ private static void validateBoundWriteMetadata(HmsTableInfo table,
ConnectorWriteHandle handle) {
+ String boundIdentity = handle.getBoundWriteMetadataIdentity();
+ if (boundIdentity != null &&
!boundIdentity.equals(writeMetadataIdentity(table))) {
+ // Bound expressions and live THiveTableSink ordinals must
describe one HMS schema generation;
+ // accepting a reorder here silently writes each value under
another column name.
+ throw new DorisConnectorException(
+ "Hive write metadata changed after the write was bound;
retry the statement");
+ }
+
+ List<ConnectorColumn> boundColumns = handle.getBoundTargetColumns();
+ int liveColumnCount = table.getColumns().size() +
table.getPartitionKeys().size();
+ if (boundColumns.isEmpty()) {
+ return;
+ }
+ if (boundColumns.size() != liveColumnCount) {
+ throw schemaChangedException();
+ }
+ for (int i = 0; i < boundColumns.size(); i++) {
+ ConnectorColumn live = i < table.getColumns().size()
+ ? table.getColumns().get(i)
+ : table.getPartitionKeys().get(i -
table.getColumns().size());
+ if
(!boundColumns.get(i).getName().equalsIgnoreCase(live.getName())) {
+ throw schemaChangedException();
+ }
+ }
+ }
+
+ private static DorisConnectorException schemaChangedException() {
+ return new DorisConnectorException(
+ "Hive table schema changed after the write was bound; retry
the statement");
+ }
+
+ static String writeMetadataIdentity(HmsTableInfo table) {
+ StringBuilder identity = new StringBuilder();
+ appendMetadataToken(identity, "data-columns");
+ for (ConnectorColumn column : table.getColumns()) {
+ appendColumnIdentity(identity, column);
+ }
+ appendMetadataToken(identity, "partition-columns");
+ for (ConnectorColumn column : table.getPartitionKeys()) {
+ appendColumnIdentity(identity, column);
+ }
+ return identity.toString();
+ }
+
+ private static void appendColumnIdentity(StringBuilder identity,
ConnectorColumn column) {
+ appendMetadataToken(identity,
column.getName().toLowerCase(Locale.ROOT));
+ appendMetadataToken(identity, column.getType());
Review Comment:
Fixed by introducing HiveWriteMetadataSnapshot, which derives the binding
columns and a fixed-size generation identity from the same table/default
inputs. The identity recursively records nested type shape and covers
SerDe/OpenCSV coercion and effective defaults. Added nested-leaf,
OpenCSV-to-LazySimple, and default-only race tests.
##########
fe/fe-connector/fe-connector-spi/src/main/java/org/apache/doris/connector/spi/ConnectorTableSchema.java:
##########
@@ -121,6 +124,18 @@ public ConnectorTableSchema(String tableName,
String tableFormatType,
Map<String, String> properties,
Set<ConnectorCapability> tableCapabilities) {
+ this(tableName, columns, tableFormatType, properties,
tableCapabilities, null);
Review Comment:
Fixed. reflectSiblingCapabilities now rebuilds the schema with the
six-argument constructor and preserves
siblingSchema.getWriteMetadataIdentity(). Added a delegated sibling regression
assertion.
##########
fe/fe-connector/fe-connector-hive/src/main/java/org/apache/doris/connector/hive/HiveWritePlanProvider.java:
##########
@@ -99,19 +100,75 @@ public ConnectorSinkPlan planWrite(ConnectorSession
session, ConnectorWriteHandl
HiveTableHandle tableHandle = (HiveTableHandle)
handle.getTableHandle();
HiveConnectorTransaction transaction = currentTransaction(session);
- // Load the table under the catalog auth context; it drives both the
location resolution
- // (buildWriteContext) and the sink assembly (buildSink). beginWrite
re-loads it for its own
- // begin-guard — the double-load is accepted (mirrors iceberg),
keeping the flow simple.
+ // One fresh table generation drives validation, transaction state,
location resolution, and sink
+ // assembly. Reloading in beginWrite would reopen a TOCTOU window
after the schema fence.
HmsTableInfo table = loadTable(tableHandle);
+ validateBoundWriteMetadata(table, handle);
Review Comment:
Fixed. The plan now carries the HMS creation metadata and full effective
metadata identity into the transaction. commit() reloads and validates that
identity before classification and again immediately before publisher setup.
Added same-shape recreation and post-plan schema-reorder tests that assert no
HMS statistics are published.
--
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]
