Fix build errors

Author: vamossagar12

fluss-flink/fluss-flink-2.2/src/main/java/org/apache/fluss/flink/catalog/Flink22Catalog.java

@@ -18,6 +18,8 @@
 
 package org.apache.fluss.flink.catalog;
 
+import org.apache.fluss.annotation.VisibleForTesting;
+import org.apache.fluss.flink.lake.LakeFlinkCatalog;
 import org.apache.fluss.metadata.TableInfo;
 
 import org.apache.flink.table.api.Schema;
@@ -31,6 +33,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
+import java.util.function.Supplier;
 
 /** A {@link FlinkCatalog} used for Flink 2.2. */
 public class Flink22Catalog extends FlinkCatalog {
@@ -40,15 +43,42 @@ public Flink22Catalog(
             String defaultDatabase,
             String bootstrapServers,
             ClassLoader classLoader,
-            Map<String, String> securityConfigs) {
-        super(name, defaultDatabase, bootstrapServers, classLoader, securityConfigs);
+            Map<String, String> securityConfigs,
+            Supplier<Map<String, String>> lakeCatalogPropertiesSupplier) {
+        super(
+                name,
+                defaultDatabase,
+                bootstrapServers,
+                classLoader,
+                securityConfigs,
+                lakeCatalogPropertiesSupplier);
+    }
+
+    @VisibleForTesting
+    public Flink22Catalog(
+            String name,
+            String defaultDatabase,
+            String bootstrapServers,
+            ClassLoader classLoader,
+            Map<String, String> securityConfigs,
+            Supplier<Map<String, String>> lakeCatalogPropertiesSupplier,
+            LakeFlinkCatalog lakeFlinkCatalog) {
+        super(
+                name,
+                defaultDatabase,
+                bootstrapServers,
+                classLoader,
+                securityConfigs,
+                lakeCatalogPropertiesSupplier,
+                lakeFlinkCatalog);
     }
 
     @Override
     public CatalogBaseTable getTable(ObjectPath objectPath)
             throws TableNotExistException, CatalogException {
         CatalogBaseTable catalogBaseTable = super.getTable(objectPath);
-        if (!(catalogBaseTable instanceof CatalogTable)) {
+        if (!(catalogBaseTable instanceof CatalogTable)
+                || objectPath.getObjectName().contains(LAKE_TABLE_SPLITTER)) {
             return catalogBaseTable;
         }
 
@@ -67,7 +97,8 @@ public CatalogBaseTable getTable(ObjectPath objectPath)
         // Judge whether we can do prefix lookup.
         TableInfo tableInfo = connection.getTable(toTablePath(objectPath)).getTableInfo();
         List<String> bucketKeys = tableInfo.getBucketKeys();
-        // For partition table, the physical primary key is the primary key that excludes the
+        // For partition table, the physical primary key is the primary key that
+        // excludes the
         // partition key
         List<String> physicalPrimaryKeys = tableInfo.getPhysicalPrimaryKeys();
         List<String> indexKeys = new ArrayList<>();

fluss-flink/fluss-flink-2.2/src/main/java/org/apache/fluss/flink/catalog/Flink22CatalogFactory.java

@@ -29,6 +29,7 @@ public FlinkCatalog createCatalog(Context context) {
                 catalog.defaultDatabase,
                 catalog.bootstrapServers,
                 catalog.classLoader,
-                catalog.securityConfigs);
+                catalog.securityConfigs,
+                catalog.lakeCatalogPropertiesSupplier);
     }
 }

fluss-flink/fluss-flink-2.2/src/test/java/org/apache/fluss/flink/catalog/Flink22CatalogITCase.java

@@ -42,7 +42,8 @@ static void beforeAll() {
                         catalog.defaultDatabase,
                         catalog.bootstrapServers,
                         catalog.classLoader,
-                        catalog.securityConfigs);
+                        catalog.securityConfigs,
+                        catalog.lakeCatalogPropertiesSupplier);
         catalog.open();
     }
 

fluss-flink/fluss-flink-2.2/src/test/java/org/apache/fluss/flink/catalog/FlinkCatalog22Test.java

@@ -0,0 +1,62 @@
+/*
+ * 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.fluss.flink.catalog;
+
+import org.apache.fluss.flink.lake.LakeFlinkCatalog;
+
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.catalog.Column;
+import org.apache.flink.table.catalog.DefaultIndex;
+import org.apache.flink.table.catalog.ResolvedSchema;
+import org.apache.flink.table.catalog.UniqueConstraint;
+
+import java.util.Arrays;
+import java.util.Collections;
+
+/** Test for {@link Flink21Catalog}. */
+public class FlinkCatalog22Test extends FlinkCatalogTest {
+
+    @Override
+    protected FlinkCatalog initCatalog(
+            String catalogName,
+            String databaseName,
+            String bootstrapServers,
+            LakeFlinkCatalog lakeFlinkCatalog) {
+        return new Flink22Catalog(
+                catalogName,
+                databaseName,
+                bootstrapServers,
+                Thread.currentThread().getContextClassLoader(),
+                Collections.emptyMap(),
+                Collections::emptyMap,
+                lakeFlinkCatalog);
+    }
+
+    protected ResolvedSchema createSchema() {
+        return new ResolvedSchema(
+                Arrays.asList(
+                        Column.physical("first", DataTypes.STRING().notNull()),
+                        Column.physical("second", DataTypes.INT()),
+                        Column.physical("third", DataTypes.STRING().notNull())),
+                Collections.emptyList(),
+                UniqueConstraint.primaryKey("PK_first_third", Arrays.asList("first", "third")),
+                Collections.singletonList(
+                        DefaultIndex.newIndex(
+                                "INDEX_first_third", Arrays.asList("first", "third"))));
+    }
+}