This is an automated email from the ASF dual-hosted git repository. amashenkov pushed a commit to branch ignite-26840 in repository https://gitbox.apache.org/repos/asf/ignite-3.git
commit 914159a7456881f4968718c638e476e7c4113bb7 Author: AMashenkov <[email protected]> AuthorDate: Thu Jan 22 21:35:13 2026 +0300 wip --- .../internal/catalog/commands/CatalogUtils.java | 22 +++- .../CatalogTableDescriptorSerializers.java | 28 ++--- .../ignite/internal/schema/SchemaDescriptor.java | 113 +++++++++++---------- .../CatalogToSchemaDescriptorConverter.java | 8 +- 4 files changed, 90 insertions(+), 81 deletions(-) diff --git a/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/commands/CatalogUtils.java b/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/commands/CatalogUtils.java index b45fbfde2fb..e02ed7994c5 100644 --- a/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/commands/CatalogUtils.java +++ b/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/commands/CatalogUtils.java @@ -61,7 +61,7 @@ import org.jetbrains.annotations.Nullable; /** * Catalog utils. */ -public class CatalogUtils { +public final class CatalogUtils { /** Default zone name. */ public static final String DEFAULT_ZONE_NAME = "Default"; @@ -967,4 +967,24 @@ public class CatalogUtils { return names; } + + /** + * Return column positions in the table descriptor for the given column IDs. + */ + public static int[] resolveColumnIndexesByIds(CatalogTableDescriptor descriptor, IntList columnIds) { + int[] columnIdxs = new int[columnIds.size()]; + + for (int i = 0; i < columnIds.size(); i++) { + int colId = columnIds.getInt(i); + int colIdx = descriptor.columnIndexById(colId); + columnIdxs[i] = colIdx; + } + + return columnIdxs; + } + + // Private constructor to prevent instantiation + private CatalogUtils() { + + } } diff --git a/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/descriptors/CatalogTableDescriptorSerializers.java b/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/descriptors/CatalogTableDescriptorSerializers.java index 7486e28da29..d881d78f433 100644 --- a/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/descriptors/CatalogTableDescriptorSerializers.java +++ b/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/descriptors/CatalogTableDescriptorSerializers.java @@ -27,6 +27,7 @@ import it.unimi.dsi.fastutil.ints.IntArrayList; import it.unimi.dsi.fastutil.ints.IntList; import java.io.IOException; import java.util.List; +import org.apache.ignite.internal.catalog.commands.CatalogUtils; import org.apache.ignite.internal.catalog.storage.serialization.CatalogEntrySerializerProvider; import org.apache.ignite.internal.catalog.storage.serialization.CatalogObjectDataInput; import org.apache.ignite.internal.catalog.storage.serialization.CatalogObjectDataOutput; @@ -125,7 +126,7 @@ public class CatalogTableDescriptorSerializers { output.writeVarInt(descriptor.primaryKeyIndexId()); output.writeVarInt(descriptor.zoneId()); - int[] pkIndexes = resolveColumnIndexesByIds(descriptor, descriptor.primaryKeyColumns()); + int[] pkIndexes = CatalogUtils.resolveColumnIndexesByIds(descriptor, descriptor.primaryKeyColumns()); output.writeVarInt(pkIndexes.length); output.writeIntArray(pkIndexes); @@ -133,7 +134,7 @@ public class CatalogTableDescriptorSerializers { if (descriptor.colocationColumns() == descriptor.primaryKeyColumns()) { output.writeVarInt(-1); } else { - int[] colocationIndexes = resolveColumnIndexesByIds(descriptor, descriptor.colocationColumns()); + int[] colocationIndexes = CatalogUtils.resolveColumnIndexesByIds(descriptor, descriptor.colocationColumns()); output.writeVarInt(colocationIndexes.length); output.writeIntArray(colocationIndexes); @@ -212,7 +213,7 @@ public class CatalogTableDescriptorSerializers { output.writeVarInt(descriptor.primaryKeyIndexId()); output.writeVarInt(descriptor.zoneId()); - int[] pkIndexes = resolveColumnIndexesByIds(descriptor, descriptor.primaryKeyColumns()); + int[] pkIndexes = CatalogUtils.resolveColumnIndexesByIds(descriptor, descriptor.primaryKeyColumns()); output.writeVarInt(pkIndexes.length); output.writeIntArray(pkIndexes); @@ -220,7 +221,7 @@ public class CatalogTableDescriptorSerializers { if (descriptor.colocationColumns() == descriptor.primaryKeyColumns()) { output.writeVarInt(-1); } else { - int[] colocationIndexes = resolveColumnIndexesByIds(descriptor, descriptor.colocationColumns()); + int[] colocationIndexes = CatalogUtils.resolveColumnIndexesByIds(descriptor, descriptor.colocationColumns()); output.writeVarInt(colocationIndexes.length); output.writeIntArray(colocationIndexes); @@ -299,7 +300,7 @@ public class CatalogTableDescriptorSerializers { output.writeVarInt(descriptor.primaryKeyIndexId()); output.writeVarInt(descriptor.zoneId()); - int[] pkIndexes = resolveColumnIndexesByIds(descriptor, descriptor.primaryKeyColumns()); + int[] pkIndexes = CatalogUtils.resolveColumnIndexesByIds(descriptor, descriptor.primaryKeyColumns()); output.writeVarInt(pkIndexes.length); output.writeIntArray(pkIndexes); @@ -307,7 +308,7 @@ public class CatalogTableDescriptorSerializers { if (descriptor.colocationColumns() == descriptor.primaryKeyColumns()) { output.writeVarInt(-1); } else { - int[] colocationIndexes = resolveColumnIndexesByIds(descriptor, descriptor.colocationColumns()); + int[] colocationIndexes = CatalogUtils.resolveColumnIndexesByIds(descriptor, descriptor.colocationColumns()); output.writeVarInt(colocationIndexes.length); output.writeIntArray(colocationIndexes); @@ -330,19 +331,4 @@ public class CatalogTableDescriptorSerializers { return columnIds; } - - /** - * Return column positions in the table descriptor for the given column IDs. - */ - private static int[] resolveColumnIndexesByIds(CatalogTableDescriptor descriptor, IntList columnIds) { - int[] columnIdxs = new int[columnIds.size()]; - - for (int i = 0; i < columnIds.size(); i++) { - int colId = columnIds.getInt(i); - int colIdx = descriptor.columnIndexById(colId); - columnIdxs[i] = colIdx; - } - - return columnIdxs; - } } diff --git a/modules/schema/src/main/java/org/apache/ignite/internal/schema/SchemaDescriptor.java b/modules/schema/src/main/java/org/apache/ignite/internal/schema/SchemaDescriptor.java index 59275d64ec2..71dd9d45a42 100644 --- a/modules/schema/src/main/java/org/apache/ignite/internal/schema/SchemaDescriptor.java +++ b/modules/schema/src/main/java/org/apache/ignite/internal/schema/SchemaDescriptor.java @@ -19,11 +19,12 @@ package org.apache.ignite.internal.schema; import static org.apache.ignite.internal.util.CollectionUtils.nullOrEmpty; +import it.unimi.dsi.fastutil.ints.IntArrayList; +import it.unimi.dsi.fastutil.ints.IntList; import it.unimi.dsi.fastutil.objects.Object2IntMap; import it.unimi.dsi.fastutil.objects.Object2IntOpenHashMap; import java.util.ArrayList; import java.util.Arrays; -import java.util.BitSet; import java.util.Collections; import java.util.HashMap; import java.util.List; @@ -93,89 +94,89 @@ public class SchemaDescriptor { } /** Constructor. */ + @TestOnly public SchemaDescriptor( int ver, List<Column> columns, List<String> keyColumns, @Nullable List<String> colocationColumns ) { + this( + ver, + columns, + IntArrayList.toList(keyColumns.stream() + .map(colName -> columns.stream().filter(c -> colName.equals(c.name())).findAny().get()) + .mapToInt(columns::indexOf)), + colocationColumns == null ? null + : IntArrayList.toList(colocationColumns.stream() + .map(colName -> columns.stream().filter(c -> colName.equals(c.name())).findAny().get()) + .mapToInt(columns::indexOf)) + ); + } + + public SchemaDescriptor(int ver, List<Column> columns, IntList keyColumnIndexes, @Nullable IntList colocationColumnIndexes) { assert !nullOrEmpty(columns) : "Schema should have at least one column"; + assert colocationColumnIndexes == null || keyColumnIndexes.containsAll(colocationColumnIndexes); + + boolean hasColocationKey = colocationColumnIndexes != null; Map<String, Column> columnsByName = new HashMap<>(); - List<Column> orderedColumns = new ArrayList<>(columns.size()); + Column[] orderedColumns = new Column[columns.size()]; + Column[] keyColumns = new Column[keyColumnIndexes.size()]; + Column[] colocationColumns = hasColocationKey ? new Column[colocationColumnIndexes.size()] : null; + Column[] valueColumns = new Column[columns.size() - keyColumnIndexes.size()]; - Object2IntMap<String> columnNameToPositionInKey = toElementToPositionMap(keyColumns); + boolean hasTemporalColumns = false; - Object2IntMap<String> columnNameToPositionInColocation; - if (colocationColumns == null) { - columnNameToPositionInColocation = columnNameToPositionInKey; - } else { - columnNameToPositionInColocation = toElementToPositionMap(colocationColumns); + IntList effectiveColocationColumns = hasColocationKey ? colocationColumnIndexes : keyColumnIndexes; - assert columnNameToPositionInKey.keySet().containsAll(colocationColumns) - : "Colocation column must be part of the key: keyCols=" - + keyColumns + ", colocationCols=" + colocationColumns; - } + for (int valueIndex = 0, rowPosition = 0; rowPosition < columns.size(); rowPosition++) { + Column column = columns.get(rowPosition); + + int keyPosition = keyColumnIndexes.indexOf(rowPosition); + int colocationPosition = effectiveColocationColumns.indexOf(rowPosition); + int valuePosition = keyPosition == -1 ? valueIndex++ : -1; - boolean hasTemporalColumns = false; - int rowPosition = 0; - int valuePosition = 0; - for (Column column : columns) { Column orderedColumn = column.copy( - rowPosition++, - columnNameToPositionInKey.getOrDefault(column.name(), -1), - columnNameToPositionInKey.containsKey(column.name()) ? -1 : valuePosition++, - columnNameToPositionInColocation.getOrDefault(column.name(), -1) + rowPosition, + keyPosition, + valuePosition, + colocationPosition ); Column old = columnsByName.put(orderedColumn.name(), orderedColumn); assert old == null : "Columns with similar names are not allowed: " + old.name(); - orderedColumns.add(orderedColumn); - - if (column.type() instanceof TemporalNativeType) { - hasTemporalColumns = true; - } - } - - this.ver = ver; - this.columns = List.copyOf(orderedColumns); - this.columnsByName = Map.copyOf(columnsByName); - this.hasTemporalColumns = hasTemporalColumns; + orderedColumns[rowPosition] = orderedColumn; - List<Column> tmpKeyColumns = new ArrayList<>(keyColumns.size()); + if (keyPosition == -1) { + assert colocationPosition == -1 : "Non key column cannot be colocation column: " + orderedColumn.name(); - BitSet keyColumnsBitSet = new BitSet(columns.size()); - for (String name : keyColumns) { - Column column = columnsByName.get(name); + valueColumns[valuePosition] = orderedColumn; + } else { + assert !orderedColumn.nullable() : "Primary key cannot contain nullable column: " + orderedColumn.name(); - assert column != null : name; - assert !column.nullable() : "Primary key cannot contain nullable column: " + name; + keyColumns[keyPosition] = orderedColumn; - tmpKeyColumns.add(column); - - assert !keyColumnsBitSet.get(column.positionInRow()) : column.name(); + if (hasColocationKey && colocationPosition != -1) { + colocationColumns[colocationPosition] = orderedColumn; + } + } - keyColumnsBitSet.set(column.positionInRow()); + hasTemporalColumns = hasTemporalColumns || (column.type() instanceof TemporalNativeType); } - this.keyCols = List.copyOf(tmpKeyColumns); - - this.colocationCols = colocationColumns == null - ? this.keyCols - : colocationColumns.stream() - .map(columnsByName::get) - .collect(Collectors.toList()); - - List<Column> tmpValueColumns = new ArrayList<>(columns.size() - keyColumnsBitSet.cardinality()); - for (Column column : orderedColumns) { - if (!keyColumnsBitSet.get(column.positionInRow())) { - tmpValueColumns.add(column); - } - } + this.ver = ver; + this.columns = List.of(orderedColumns); + this.columnsByName = Map.copyOf(columnsByName); + this.hasTemporalColumns = hasTemporalColumns; - this.valCols = List.copyOf(tmpValueColumns); + this.keyCols = List.of(keyColumns); + this.valCols = List.of(valueColumns); + this.colocationCols = hasColocationKey + ? List.of(colocationColumns) + : this.keyCols; } private static List<Column> mergeColumns(Column[] keyColumns, Column[] valueColumns) { diff --git a/modules/schema/src/main/java/org/apache/ignite/internal/schema/catalog/CatalogToSchemaDescriptorConverter.java b/modules/schema/src/main/java/org/apache/ignite/internal/schema/catalog/CatalogToSchemaDescriptorConverter.java index c4780c26b21..409afa99b62 100644 --- a/modules/schema/src/main/java/org/apache/ignite/internal/schema/catalog/CatalogToSchemaDescriptorConverter.java +++ b/modules/schema/src/main/java/org/apache/ignite/internal/schema/catalog/CatalogToSchemaDescriptorConverter.java @@ -19,6 +19,7 @@ package org.apache.ignite.internal.schema.catalog; import static org.apache.ignite.internal.lang.IgniteStringFormatter.format; +import it.unimi.dsi.fastutil.ints.IntList; import java.util.ArrayList; import java.util.HashMap; import java.util.List; @@ -172,13 +173,14 @@ public final class CatalogToSchemaDescriptorConverter { columns.add(convert(column)); } - List<String> colocationColumns = tableDescriptor.primaryKeyColumns() == tableDescriptor.colocationColumns() - ? null : CatalogUtils.resolveColumnNames(tableDescriptor, tableDescriptor.colocationColumns()); + IntList pkColumns = IntList.of(CatalogUtils.resolveColumnIndexesByIds(tableDescriptor, tableDescriptor.primaryKeyColumns())); + IntList colocationColumns = tableDescriptor.primaryKeyColumns() == tableDescriptor.colocationColumns() + ? null : IntList.of(CatalogUtils.resolveColumnIndexesByIds(tableDescriptor, tableDescriptor.colocationColumns())); return new SchemaDescriptor( tableVersion, columns, - CatalogUtils.resolveColumnNames(tableDescriptor, tableDescriptor.primaryKeyColumns()), + pkColumns, colocationColumns ); }
