FrankChen021 commented on code in PR #19830:
URL: https://github.com/apache/druid/pull/19830#discussion_r3775307632
##########
extensions-core/druid-catalog/src/main/java/org/apache/druid/catalog/sync/CatalogClient.java:
##########
@@ -105,23 +107,58 @@ public ResolvedTable resolveTable(TableId id)
/**
* Creates a table for the given {@link TableId} and {@link TableSpec}.
* If a table already exists for this id, it is overwritten.
- * <p>
- * This method is currently used only in tests.
*/
public void createTable(TableId tableId, TableSpec tableSpec)
{
- getResult(postCreateTable(tableId, tableSpec));
+ getResult(postCreateTable(tableId, tableSpec, false, true));
+ }
+
+ /**
+ * Creates a table for the given {@link TableId} and {@link TableSpec}.
+ *
+ * @param ifNotExists leave an existing table alone rather than failing
+ * @param overwrite replace the spec of an existing table
+ */
+ public void createTable(TableId tableId, TableSpec tableSpec, boolean
ifNotExists, boolean overwrite)
+ {
+ FutureUtils.getUnchecked(postCreateTable(tableId, tableSpec, ifNotExists,
overwrite), true);
+ }
+
+ /**
+ * Applies an edit to an existing table's catalog entry.
+ * <p>
+ * API: {@code POST
/druid/coordinator/v1/catalog/schemas/{schema}/tables/{name}/edit}
+ */
+ public void editTable(TableId tableId, TableEditRequest editRequest)
+ {
+ String path = tablePath(TABLE_EDIT_PATH, tableId);
+ FutureUtils.getUnchecked(
+ serviceClient.asyncRequest(
Review Comment:
[P1] Retries can replay catalog mutations
This new mutating POST goes through a client configured with
`StandardRetryPolicy.maxAttempts(6)`. If the Coordinator commits the DDL but
the response is lost, a retryable channel error or 5xx can replay the
non-idempotent edit (and the analogous create request), producing conflicts or
duplicate effects while the SQL caller sees failure. Disable retries for writes
or add request idempotency/deduplication.
##########
extensions-core/druid-catalog/src/main/java/org/apache/druid/catalog/storage/sql/SQLCatalogManager.java:
##########
@@ -453,44 +436,44 @@ public TableMetadata withHandle(Handle handle) throws
CatalogException
{
handle.begin();
try {
- final Query<Map<String, Object>> query = handle
- .createQuery(statement(SELECT_COLUMNS_STMT))
+ final ResultIterator<TableMetadata> resultIterator = handle
+ .createQuery(statement(selectStmt))
.setFetchSize(connector.getStreamingFetchSize())
.bind(SCHEMA_NAME_COL, id.schema())
- .bind(TABLE_NAME_COL, id.name());
-
- final ResultIterator<TableSpec> resultIterator = query
- .map((index, r, ctx) ->
- tableSpecFromBytes(
- jsonMapper,
- r.getString(1),
- null,
- r.getBytes(2)
- )
- )
- .iterator();
- final TableSpec tableSpec;
+ .bind(TABLE_NAME_COL, id.name())
+ .map((index, r, ctx) ->
+ TableMetadata.forUpdate(
+ id,
+ r.getLong(4),
+ tableSpecFromBytes(jsonMapper, r.getString(1),
r.getBytes(2), r.getBytes(3))
+ )
+ )
+ .iterator();
+ final TableMetadata existing;
if (resultIterator.hasNext()) {
- tableSpec = resultIterator.next();
+ existing = resultIterator.next();
} else {
throw tableNotFound(id);
}
- final TableSpec revised = transform.apply(TableMetadata.of(id,
tableSpec));
+ final TableSpec revised = transform.apply(existing);
if (revised == null) {
handle.rollback();
return null;
}
final long updateTime = System.currentTimeMillis();
final int updateCount = handle
- .createStatement(statement(UPDATE_COLUMNS_STMT))
+ .createStatement(statement(updateStmt))
.bind(SCHEMA_NAME_COL, id.schema())
.bind(TABLE_NAME_COL, id.name())
- .bind(COLUMNS_COL, JacksonUtils.toBytes(jsonMapper,
revised.columns()))
+ .bind(blobColumn, JacksonUtils.toBytes(jsonMapper,
blob.apply(revised)))
.bind(UPDATE_TIME_COL, updateTime)
+ .bind(OLD_VERSION_PARAM, existing.updateTime())
Review Comment:
[P1] CAS version can repeat
The new compare-and-set update uses `System.currentTimeMillis()` as the
replacement version, while the predicate compares the row against
`existing.updateTime()`. Two edits in the same millisecond can therefore reuse
the same version: both transactions may pass the old-version predicate,
overwrite one another, and publish indistinguishable cache versions. Generate a
strictly increasing version atomically (or use an equivalent database-side
mechanism) before relying on this CAS.
--
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]
