Support primary key columns ordering for standard pipeline table metadata loader (#32647)

Author: sandynz

kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/consistencycheck/DataConsistencyCheckUtils.java

@@ -30,6 +30,7 @@
 import java.sql.Array;
 import java.sql.SQLException;
 import java.sql.SQLXML;
+import java.util.Collection;
 import java.util.Iterator;
 import java.util.Map;
 import java.util.Map.Entry;
@@ -150,6 +151,33 @@ public static boolean isBigDecimalEquals(final BigDecimal one, final BigDecimal
         return 0 == decimalOne.compareTo(decimalTwo);
     }
 
+    /**
+     * Compare lists.
+     *
+     * @param thisList this list
+     * @param thatList that list
+     * @return true if lists equals, otherwise false
+     */
+    public static boolean compareLists(final @Nullable Collection<?> thisList, final @Nullable Collection<?> thatList) {
+        if (null == thisList && null == thatList) {
+            return true;
+        }
+        if (null == thisList || null == thatList) {
+            return false;
+        }
+        if (thisList.size() != thatList.size()) {
+            return false;
+        }
+        Iterator<?> thisIterator = thisList.iterator();
+        Iterator<?> thatIterator = thatList.iterator();
+        while (thisIterator.hasNext() && thatIterator.hasNext()) {
+            if (!Objects.deepEquals(thisIterator.next(), thatIterator.next())) {
+                return false;
+            }
+        }
+        return true;
+    }
+    
     /**
      * Get first unique key value.
      *

kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/metadata/loader/StandardPipelineTableMetaDataLoader.java

@@ -19,6 +19,7 @@
 
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
+import org.apache.shardingsphere.data.pipeline.core.consistencycheck.DataConsistencyCheckUtils;
 import org.apache.shardingsphere.data.pipeline.core.datasource.PipelineDataSourceWrapper;
 import org.apache.shardingsphere.data.pipeline.core.exception.PipelineInternalException;
 import org.apache.shardingsphere.data.pipeline.core.metadata.model.PipelineColumnMetaData;
@@ -33,11 +34,9 @@
 import java.sql.SQLException;
 import java.util.Collection;
 import java.util.LinkedHashMap;
-import java.util.LinkedHashSet;
 import java.util.LinkedList;
 import java.util.Map;
 import java.util.Map.Entry;
-import java.util.Set;
 import java.util.SortedMap;
 import java.util.TreeMap;
 import java.util.concurrent.ConcurrentHashMap;
@@ -91,7 +90,7 @@ private Map<CaseInsensitiveIdentifier, PipelineTableMetaData> loadTableMetaData0
         }
         Map<CaseInsensitiveIdentifier, PipelineTableMetaData> result = new LinkedHashMap<>(tableNames.size(), 1F);
         for (String each : tableNames) {
-            Set<CaseInsensitiveIdentifier> primaryKeys = loadPrimaryKeys(connection, schemaName, each);
+            Collection<CaseInsensitiveIdentifier> primaryKeys = loadPrimaryKeys(connection, schemaName, each);
             Map<CaseInsensitiveIdentifier, Collection<CaseInsensitiveIdentifier>> uniqueKeys = loadUniqueIndexesOfTable(connection, schemaName, each);
             Map<CaseInsensitiveIdentifier, PipelineColumnMetaData> columnMetaDataMap = new LinkedHashMap<>();
             try (ResultSet resultSet = connection.getMetaData().getColumns(connection.getCatalog(), schemaName, each, "%")) {
@@ -111,7 +110,9 @@ private Map<CaseInsensitiveIdentifier, PipelineTableMetaData> loadTableMetaData0
                 }
             }
             Collection<PipelineIndexMetaData> uniqueIndexMetaData = uniqueKeys.entrySet().stream()
-                    .map(entry -> new PipelineIndexMetaData(entry.getKey(), entry.getValue().stream().map(columnMetaDataMap::get).collect(Collectors.toList()))).collect(Collectors.toList());
+                    .map(entry -> new PipelineIndexMetaData(entry.getKey(), entry.getValue().stream().map(columnMetaDataMap::get).collect(Collectors.toList()),
+                            DataConsistencyCheckUtils.compareLists(primaryKeys, entry.getValue())))
+                    .collect(Collectors.toList());
             result.put(new CaseInsensitiveIdentifier(each), new PipelineTableMetaData(each, columnMetaDataMap, uniqueIndexMetaData));
         }
         return result;
@@ -138,14 +139,13 @@ private Map<CaseInsensitiveIdentifier, Collection<CaseInsensitiveIdentifier>> lo
         return result;
     }
 
-    private Set<CaseInsensitiveIdentifier> loadPrimaryKeys(final Connection connection, final String schemaName, final String tableName) throws SQLException {
-        Set<CaseInsensitiveIdentifier> result = new LinkedHashSet<>();
-        // TODO order primary keys
+    private Collection<CaseInsensitiveIdentifier> loadPrimaryKeys(final Connection connection, final String schemaName, final String tableName) throws SQLException {
+        SortedMap<Short, CaseInsensitiveIdentifier> result = new TreeMap<>();
         try (ResultSet resultSet = connection.getMetaData().getPrimaryKeys(connection.getCatalog(), schemaName, tableName)) {
             while (resultSet.next()) {
-                result.add(new CaseInsensitiveIdentifier(resultSet.getString("COLUMN_NAME")));
+                result.put(resultSet.getShort("KEY_SEQ"), new CaseInsensitiveIdentifier(resultSet.getString("COLUMN_NAME")));
             }
         }
-        return result;
+        return result.values();
     }
 }

kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/metadata/model/PipelineIndexMetaData.java

@@ -35,4 +35,6 @@ public final class PipelineIndexMetaData {
     private final CaseInsensitiveIdentifier name;
 
     private final List<PipelineColumnMetaData> columns;
+    
+    private final boolean primaryKey;
 }

kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/metadata/model/PipelineTableMetaData.java

@@ -31,6 +31,7 @@
 import java.util.Collections;
 import java.util.List;
 import java.util.Map;
+import java.util.Optional;
 import java.util.stream.Collectors;
 
 /**
@@ -62,8 +63,10 @@ public PipelineTableMetaData(final String name, final Map<CaseInsensitiveIdentif
         List<PipelineColumnMetaData> columnMetaDataList = new ArrayList<>(columnMetaDataMap.values());
         Collections.sort(columnMetaDataList);
         columnNames = Collections.unmodifiableList(columnMetaDataList.stream().map(PipelineColumnMetaData::getName).collect(Collectors.toList()));
-        primaryKeyColumns = Collections.unmodifiableList(columnMetaDataList.stream().filter(PipelineColumnMetaData::isPrimaryKey)
-                .map(PipelineColumnMetaData::getName).collect(Collectors.toList()));
+        Optional<PipelineIndexMetaData> primaryKeyMetaData = uniqueIndexes.stream().filter(PipelineIndexMetaData::isPrimaryKey).findFirst();
+        primaryKeyColumns = primaryKeyMetaData.map(each -> each.getColumns().stream().map(PipelineColumnMetaData::getName).collect(Collectors.toList()))
+                .orElseGet(() -> Collections.unmodifiableList(columnMetaDataList.stream().filter(PipelineColumnMetaData::isPrimaryKey)
+                        .map(PipelineColumnMetaData::getName).collect(Collectors.toList())));
         this.uniqueIndexes = Collections.unmodifiableCollection(uniqueIndexes);
     }
 

kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/util/PipelineStringUtils.java

@@ -0,0 +1,58 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.shardingsphere.data.pipeline.core.util;
+
+import lombok.AccessLevel;
+import lombok.NoArgsConstructor;
+
+import java.util.Collection;
+import java.util.Iterator;
+
+/**
+ * Pipeline string utils.
+ */
+@NoArgsConstructor(access = AccessLevel.NONE)
+public final class PipelineStringUtils {
+    
+    /**
+     * Equals ignore case.
+     *
+     * @param one one
+     * @param another another
+     * @return is equals ignore case
+     */
+    public static boolean equalsIgnoreCase(final Collection<String> one, final Collection<String> another) {
+        if (null == one && null == another) {
+            return true;
+        }
+        if (null == one || null == another) {
+            return false;
+        }
+        if (one.size() != another.size()) {
+            return false;
+        }
+        Iterator<String> oneIterator = one.iterator();
+        Iterator<String> anotherIterator = another.iterator();
+        while (oneIterator.hasNext() && anotherIterator.hasNext()) {
+            if (!oneIterator.next().equalsIgnoreCase(anotherIterator.next())) {
+                return false;
+            }
+        }
+        return true;
+    }
+}

kernel/data-pipeline/core/src/test/java/org/apache/shardingsphere/data/pipeline/core/util/PipelineStringUtilsTest.java

@@ -0,0 +1,40 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.shardingsphere.data.pipeline.core.util;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.Collections;
+
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+class PipelineStringUtilsTest {
+    
+    @Test
+    void assertEqualsIgnoreCase() {
+        assertTrue(PipelineStringUtils.equalsIgnoreCase(null, null));
+        assertFalse(PipelineStringUtils.equalsIgnoreCase(null, Collections.emptyList()));
+        assertFalse(PipelineStringUtils.equalsIgnoreCase(Collections.emptyList(), null));
+        assertFalse(PipelineStringUtils.equalsIgnoreCase(Collections.singletonList("test"), Collections.emptyList()));
+        assertFalse(PipelineStringUtils.equalsIgnoreCase(Collections.emptyList(), Collections.singletonList("test")));
+        assertTrue(PipelineStringUtils.equalsIgnoreCase(Collections.singletonList("test"), Collections.singletonList("test")));
+        assertTrue(PipelineStringUtils.equalsIgnoreCase(Collections.singletonList("TEST"), Collections.singletonList("test")));
+        assertTrue(PipelineStringUtils.equalsIgnoreCase(Collections.singletonList("test"), Collections.singletonList("TEST")));
+    }
+}