github-actions[bot] commented on code in PR #66307:
URL: https://github.com/apache/doris/pull/66307#discussion_r3820353117
##########
fe/fe-core/src/main/java/org/apache/doris/catalog/constraint/ConstraintManager.java:
##########
@@ -980,17 +1369,25 @@ private void
swapPrimaryKeyForeignTables(PrimaryKeyConstraint pk,
// ==================== EditLog integration ====================
- private void logAddConstraint(TableNameInfo tableNameInfo,
+ private EditLog.EditLogItem submitAddConstraint(TableNameInfo
tableNameInfo,
Constraint constraint) {
AlterConstraintLog log = new AlterConstraintLog(
constraint, tableNameInfo);
- Env.getCurrentEnv().getEditLog().logAddConstraint(log);
+ return Env.getCurrentEnv().getEditLog()
+ .submitEdit(OperationType.OP_ADD_CONSTRAINT, log);
Review Comment:
[P1] Snapshot constraint logs before releasing the locks
`AlterConstraintLog` keeps this same mutable `Constraint` object, and
`submitEdit` only queues the `Writable`; Gson serialization happens later in
`JournalBatch.addJournal` on the flusher. After a PK ADD releases its
table/manager locks but before that queued payload is serialized, an FK ADD can
mutate the PK via `addForeignTable`. The earlier PK journal record then
includes a reverse reference created by the later operation. With non-batch
flushing, a crash after the PK write but before the FK write replays a ghost
reference to an FK that never became durable, so subsequent PK/table drops can
be rejected incorrectly. FIFO ordering does not snapshot mutable payloads.
Please serialize/deep-copy the log while the metadata locks are held, enqueue
that immutable snapshot, and add a paused-flusher crash-prefix replay test.
##########
fe/fe-core/src/main/java/org/apache/doris/catalog/constraint/ConstraintManager.java:
##########
@@ -893,6 +1172,116 @@ private void validateTableAndColumns(TableNameInfo
tableNameInfo,
fk.getReferencedColumnNames(),
toKey(refTableInfo));
}
+ } else if (constraint instanceof DistributionMappingConstraint) {
+ validateDistributionMappingConstraint(
+ tableNameInfo, table, (DistributionMappingConstraint)
constraint);
+ }
+ return table;
+ }
+
+ private void validateResolvedConstraint(TableNameInfo tableNameInfo,
TableIf table,
+ TableIf referencedTable, Constraint constraint) {
+ if (constraint instanceof PrimaryKeyConstraint) {
+ validateColumnsExist(table,
+ ((PrimaryKeyConstraint) constraint).getPrimaryKeyNames(),
+ toKey(tableNameInfo));
+ } else if (constraint instanceof UniqueConstraint) {
+ validateColumnsExist(table,
+ ((UniqueConstraint) constraint).getUniqueColumnNames(),
+ toKey(tableNameInfo));
+ } else if (constraint instanceof ForeignKeyConstraint) {
+ if (referencedTable == null) {
+ throw new AnalysisException("Referenced table changed while
adding constraint on "
+ + tableNameInfo);
+ }
+ ForeignKeyConstraint foreignKey = (ForeignKeyConstraint)
constraint;
+ validateColumnsExist(table, foreignKey.getForeignKeyNames(),
toKey(tableNameInfo));
+ validateColumnsExist(referencedTable,
foreignKey.getReferencedColumnNames(),
+ toKey(foreignKey.getReferencedTableName()));
+ } else if (constraint instanceof DistributionMappingConstraint) {
+ validateDistributionMappingConstraint(
+ tableNameInfo, table, (DistributionMappingConstraint)
constraint);
+ }
+ }
+
+ private TableIf resolveTableIfPresent(TableNameInfo tableNameInfo) {
+ try {
+ return resolveTableForValidation(tableNameInfo);
+ } catch (AnalysisException e) {
+ LOG.debug("Table {} is unavailable while synchronizing table-local
constraints",
+ tableNameInfo, e);
+ return null;
+ }
+ }
+
+ @SuppressWarnings("deprecation")
+ private void putTableLocalConstraint(TableIf table, String constraintName,
Constraint constraint) {
+ if (table instanceof Table) {
+ ((Table)
table).getTableAttributes().getConstraintsMap().put(constraintName, constraint);
+ }
+ }
+
+ @SuppressWarnings("deprecation")
+ private void removeTableLocalConstraint(TableIf table, String
constraintName) {
+ if (table instanceof Table) {
+ ((Table)
table).getTableAttributes().getConstraintsMap().remove(constraintName);
+ }
+ }
+
+ private void validateDistributionMappingConstraint(TableNameInfo
tableNameInfo, TableIf table,
+ DistributionMappingConstraint constraint) {
+ if (!(table instanceof OlapTable)) {
+ throw new AnalysisException("Distribution mapping constraint only
supports OLAP tables");
+ }
+ validateColumnsExist(table, constraint.getDeterminantColumnNames(),
toKey(tableNameInfo));
+ validateColumnsExist(table, constraint.getDistributionColumnNames(),
toKey(tableNameInfo));
+ TreeSet<String> determinantColumns = new
TreeSet<>(String.CASE_INSENSITIVE_ORDER);
+ determinantColumns.addAll(constraint.getDeterminantColumnNames());
+ if (determinantColumns.size() !=
constraint.getDeterminantColumnNames().size()) {
+ throw new AnalysisException("Determinant columns in distribution
mapping constraint must be unique");
+ }
+ TreeSet<String> distributionColumns = new
TreeSet<>(String.CASE_INSENSITIVE_ORDER);
+ distributionColumns.addAll(constraint.getDistributionColumnNames());
+ if (distributionColumns.size() !=
constraint.getDistributionColumnNames().size()) {
+ throw new AnalysisException("Distribution columns in distribution
mapping constraint must be unique");
+ }
+
+ OlapTable olapTable = (OlapTable) table;
+ if (!(olapTable.getDefaultDistributionInfo() instanceof
HashDistributionInfo)) {
+ throw new AnalysisException("Distribution mapping constraint
requires hash distribution");
+ }
+ List<String> tableDistributionColumns = ((HashDistributionInfo)
olapTable.getDefaultDistributionInfo())
+ .getDistributionColumns().stream()
+ .map(column -> column.getName().toLowerCase(Locale.ROOT))
+ .collect(Collectors.toList());
+ List<String> constraintDistributionColumns =
constraint.getDistributionColumnNames().stream()
+ .map(column -> column.toLowerCase(Locale.ROOT))
+ .collect(Collectors.toList());
+ int previousIndex = -1;
+ for (String column : constraintDistributionColumns) {
+ int index = tableDistributionColumns.indexOf(column);
+ if (index <= previousIndex) {
+ throw new AnalysisException("Distribution columns in
distribution mapping constraint"
+ + " must be an ordered subset of table distribution
columns");
+ }
+ previousIndex = index;
+ }
+ }
+
+ private void validateFrontendVersionsForDistributionMappingConstraint() {
+ String currentVersion = Version.DORIS_BUILD_VERSION + "-" +
Version.DORIS_BUILD_SHORT_HASH;
+ List<String> incompatibleFrontends = new ArrayList<>();
+ for (Frontend frontend : Env.getCurrentEnv().getFrontends(null)) {
Review Comment:
[P2] Fence frontend admission with the subtype check
This snapshots the frontend map under the ConstraintManager lock, but `ADD
FRONTEND` uses the unrelated Env lock. An older follower can be inserted just
after this loop and its `OP_ADD_FRONTEND` can be queued before or after
`OP_ADD_CONSTRAINT`; in either order it eventually loads a
`DistributionMappingConstraint` that its Gson factory does not know and fails
image/edit-log deserialization. Later old-FE admission is unguarded for the
same reason, and the candidate cannot report `Frontend.version` until after it
has become ready, which is already after image loading/replay. The PR-body
warning documents this failure but does not make the two metadata operations
safe. Please use one capability/admission fence (or a backward-readable
encoding) and add a latch-based concurrent ADD plus later-admission replay test.
##########
fe/fe-core/src/main/java/org/apache/doris/catalog/constraint/ConstraintManager.java:
##########
@@ -361,40 +617,53 @@ private void dropConstraintsByPrefix(String prefix) {
*/
public void renameTable(TableNameInfo oldTableInfo,
TableNameInfo newTableInfo) {
- String oldKey = toKey(oldTableInfo);
- String newKey = toKey(newTableInfo);
writeLock();
try {
- // Move this table's own constraints
- Map<String, Constraint> tableConstraints
- = constraintsMap.remove(oldKey);
- if (tableConstraints != null) {
- constraintsMap.put(newKey, tableConstraints);
+ renameTableWithoutLock(oldTableInfo, newTableInfo);
+ } finally {
+ writeUnlock();
+ }
+ }
+
+ /** Move every qualified table key when a database is renamed. */
+ public void renameDatabase(String catalogName, String oldDbName, String
newDbName) {
+ String oldPrefix = catalogName + "." + oldDbName + ".";
+ writeLock();
+ try {
+ List<TableNameInfo> oldTableInfos =
constraintsMap.keySet().stream()
+ .filter(key -> key.startsWith(oldPrefix))
+ .map(TableNameInfo::new)
+ .collect(Collectors.toList());
+ for (TableNameInfo oldTableInfo : oldTableInfos) {
Review Comment:
[P2] Rewrite database constraints in one pass
This invokes `renameTableWithoutLock` once for every constrained table in
the database, but each invocation rescans every table's constraints and every
FK/PK reference. Renaming a database with M constrained tables in a catalog
containing N constraints is therefore O(M*N) (quadratic when most constraints
are in that database), while the internal-catalog, database, and
ConstraintManager write locks are all held and before the rename journal
completes. Large constraint catalogs can stall metadata operations and replay.
Please move all affected keys in one pass and rewrite references with a single
traversal (and avoid per-table INFO logging), with a many-table scale test.
--
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]
