Disable the Wren Protocol and cache layer in default (#546)

* disable pg wire protocol and cache in default

* enable the pg wire protocol for testing

* fix test injection failed

* fix dependency issue

* fix tests

wren-base/src/main/java/io/wren/base/config/PostgresWireProtocolConfig.java

@@ -20,6 +20,7 @@
 
 public class PostgresWireProtocolConfig
 {
+    public static final String PG_WIRE_PROTOCOL_ENABLED = "pg-wire-protocol.enabled";
     public static final String PG_WIRE_PROTOCOL_SSL_ENABLED = "pg-wire-protocol.ssl.enabled";
     public static final String PG_WIRE_PROTOCOL_NETTY_THREAD_COUNT = "pg-wire-protocol.netty.thread.count";
     public static final String PG_WIRE_PROTOCOL_AUTH_FILE = "pg-wire-protocol.auth.file";
@@ -29,6 +30,7 @@ public class PostgresWireProtocolConfig
     private boolean sslEnable;
     private int nettyThreadCount;
     private File authFile = new File("etc/accounts");
+    private boolean pgWireProtocolEnabled;
 
     @NotNull
     public String getPort()
@@ -80,4 +82,15 @@ public PostgresWireProtocolConfig setAuthFile(File authFile)
         this.authFile = authFile;
         return this;
     }
+
+    @Config(PG_WIRE_PROTOCOL_ENABLED)
+    public void setPgWireProtocolEnabled(boolean pgWireProtocolEnabled)
+    {
+        this.pgWireProtocolEnabled = pgWireProtocolEnabled;
+    }
+
+    public boolean isPgWireProtocolEnabled()
+    {
+        return pgWireProtocolEnabled;
+    }
 }

wren-cache/src/main/java/io/wren/cache/CacheModule.java

@@ -17,20 +17,14 @@
 import com.google.inject.Binder;
 import com.google.inject.Scopes;
 import io.airlift.configuration.AbstractConfigurationAwareModule;
-import io.wren.base.client.duckdb.DuckDBConfig;
-import io.wren.base.client.duckdb.DuckdbS3StyleStorageConfig;
-
-import static io.airlift.configuration.ConfigBinder.configBinder;
 
 public class CacheModule
         extends AbstractConfigurationAwareModule
 {
     @Override
     protected void setup(Binder binder)
     {
-        configBinder(binder).bindConfig(DuckdbS3StyleStorageConfig.class);
-        configBinder(binder).bindConfig(DuckDBConfig.class);
-        binder.bind(CacheManager.class).in(Scopes.SINGLETON);
+        binder.bind(CacheManager.class).to(CacheManagerImpl.class).in(Scopes.SINGLETON);
         binder.bind(CacheTaskManager.class).in(Scopes.SINGLETON);
         binder.bind(EventLogger.class).to(Log4jEventLogger.class).in(Scopes.SINGLETON);
         binder.bind(CachedTableMapping.class).to(DefaultCachedTableMapping.class).in(Scopes.SINGLETON);

wren-cache/src/main/java/io/wren/cache/NoOpCacheManager.java

@@ -0,0 +1,20 @@
+/*
+ * Licensed 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 io.wren.cache;
+
+public class NoOpCacheManager
+        implements CacheManager
+{
+}

wren-main/src/main/java/io/wren/main/WrenModule.java

@@ -17,7 +17,14 @@
 import com.google.inject.Binder;
 import com.google.inject.Scopes;
 import io.airlift.configuration.AbstractConfigurationAwareModule;
+import io.wren.base.config.PostgresWireProtocolConfig;
 import io.wren.base.config.WrenConfig;
+import io.wren.cache.CacheManager;
+import io.wren.cache.CacheManagerImpl;
+import io.wren.cache.NoOpCacheManager;
+import io.wren.main.pgcatalog.NoOpPgCatalogManager;
+import io.wren.main.pgcatalog.PgCatalogManager;
+import io.wren.main.pgcatalog.PgCatalogManagerImpl;
 
 import static io.airlift.configuration.ConfigBinder.configBinder;
 
@@ -27,8 +34,17 @@ public class WrenModule
     @Override
     protected void setup(Binder binder)
     {
+        PostgresWireProtocolConfig config = buildConfigObject(PostgresWireProtocolConfig.class);
         configBinder(binder).bindConfig(WrenConfig.class);
         binder.bind(WrenManager.class).in(Scopes.SINGLETON);
         binder.bind(WrenMetastore.class).in(Scopes.SINGLETON);
+        if (config.isPgWireProtocolEnabled()) {
+            binder.bind(CacheManager.class).to(CacheManagerImpl.class).in(Scopes.SINGLETON);
+            binder.bind(PgCatalogManager.class).to(PgCatalogManagerImpl.class).in(Scopes.SINGLETON);
+        }
+        else {
+            binder.bind(CacheManager.class).to(NoOpCacheManager.class).in(Scopes.SINGLETON);
+            binder.bind(PgCatalogManager.class).to(NoOpPgCatalogManager.class).in(Scopes.SINGLETON);
+        }
     }
 }

wren-main/src/main/java/io/wren/main/pgcatalog/NoOpPgCatalogManager.java

@@ -0,0 +1,20 @@
+/*
+ * Licensed 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 io.wren.main.pgcatalog;
+
+public class NoOpPgCatalogManager
+        implements PgCatalogManager
+{
+}

wren-main/src/main/java/io/wren/main/pgcatalog/PgCatalogManager.java

@@ -14,177 +14,24 @@
 
 package io.wren.main.pgcatalog;
 
-import com.google.inject.Inject;
-import io.airlift.log.Logger;
-import io.wren.base.Column;
-import io.wren.base.WrenMDL;
-import io.wren.base.pgcatalog.function.DataSourceFunctionRegistry;
-import io.wren.base.pgcatalog.function.PgMetastoreFunctionRegistry;
-import io.wren.base.wireprotocol.PgMetastore;
-import io.wren.main.PreviewService;
-import io.wren.main.WrenMetastore;
-import io.wren.main.metadata.Metadata;
-import io.wren.main.pgcatalog.builder.PgFunctionBuilderManager;
-import io.wren.main.pgcatalog.builder.PgMetastoreFunctionBuilder;
 import io.wren.main.pgcatalog.exception.DeployException;
-import org.apache.commons.lang3.StringUtils;
 
-import java.util.List;
-import java.util.Optional;
-
-import static java.lang.String.format;
-import static java.util.Objects.requireNonNull;
-import static java.util.stream.Collectors.joining;
-
-public class PgCatalogManager
+public interface PgCatalogManager
 {
-    private static final Logger LOG = Logger.get(PgCatalogManager.class);
-
-    protected final String pgCatalogName;
-
-    private final Metadata connector;
-    private final DataSourceFunctionRegistry dataSourceFunctionRegistry;
-    private final PgMetastoreFunctionRegistry metastoreFunctionRegistry;
-    private final PgFunctionBuilderManager pgFunctionBuilderManager;
-    private final PgMetastoreFunctionBuilder pgMetastoreFunctionBuilder;
-    private final PgMetastore pgMetastore;
-    private final WrenMetastore wrenMetastore;
-    private final PreviewService previewService;
-
-    @Inject
-    public PgCatalogManager(
-            Metadata connector,
-            PgFunctionBuilderManager pgFunctionBuilderManager,
-            PgMetastore pgMetastore,
-            WrenMetastore wrenMetastore,
-            PreviewService previewService)
-    {
-        this.connector = requireNonNull(connector, "connector is null");
-        this.pgFunctionBuilderManager = requireNonNull(pgFunctionBuilderManager, "pgFunctionBuilderManager is null");
-        this.pgCatalogName = requireNonNull(connector.getPgCatalogName());
-        this.dataSourceFunctionRegistry = new DataSourceFunctionRegistry();
-        this.metastoreFunctionRegistry = new PgMetastoreFunctionRegistry();
-        this.pgMetastore = requireNonNull(pgMetastore, "pgMetastore is null");
-        this.pgMetastoreFunctionBuilder = new PgMetastoreFunctionBuilder(pgMetastore);
-        this.wrenMetastore = requireNonNull(wrenMetastore, "wrenMetastore is null");
-        this.previewService = requireNonNull(previewService, "previewService is null");
-    }
-
-    public void initPgCatalog()
+    default void initPgCatalog()
             throws DeployException
-    {
-        if (!connector.isPgCompatible()) {
-            createOrReplaceSchema(pgCatalogName);
-        }
-        initPgFunctions();
-        syncPgMetastore();
-    }
+    {}
 
-    public void initPgFunctions()
-    {
-        metastoreFunctionRegistry.getFunctions()
-                .stream()
-                .filter(f -> !f.isImplemented())
-                .forEach(pgMetastoreFunctionBuilder::createPgFunction);
-        if (!connector.isPgCompatible()) {
-            dataSourceFunctionRegistry.getFunctions()
-                    .stream()
-                    .filter(f -> !f.isImplemented())
-                    .forEach(pgFunctionBuilderManager::createPgFunction);
-        }
-    }
+    default void initPgFunctions() {}
 
-    public void dropSchema(String name)
-    {
-        pgMetastore.directDDL(format("DROP SCHEMA IF EXISTS \"%s\" CASCADE;", name));
-    }
-
-    private void createOrReplaceSchema(String name)
-    {
-        connector.dropSchemaIfExists(name);
-        connector.createSchema(name);
-    }
+    default void dropSchema(String name) {}
 
-    public boolean checkRequired()
+    default boolean checkRequired()
     {
-        WrenMDL mdl = wrenMetastore.getAnalyzedMDL().getWrenMDL();
-        if (!(pgMetastore.isSchemaExist(mdl.getSchema()))) {
-            LOG.warn("PgCatalog is not initialized");
-            return false;
-        }
         return true;
     }
 
-    public void syncPgMetastore()
+    default void syncPgMetastore()
             throws DeployException
-    {
-        try {
-            WrenMDL mdl = wrenMetastore.getAnalyzedMDL().getWrenMDL();
-            StringBuilder sb = new StringBuilder();
-            if (StringUtils.isNotEmpty(mdl.getSchema())) {
-                sb.append(format("DROP SCHEMA IF EXISTS \"%s\" CASCADE;\n", mdl.getSchema()));
-                sb.append(format("CREATE SCHEMA IF NOT EXISTS \"%s\";\n", mdl.getSchema()));
-            }
-            mdl.listModels().forEach(model -> {
-                String cols = model.getColumns().stream()
-                        .filter(column -> column.getRelationship().isEmpty())
-                        .map(column -> format("\"%s\" %s", column.getName(), pgMetastore.handlePgType(column.getType())))
-                        .collect(joining(","));
-                sb.append(format("CREATE TABLE IF NOT EXISTS \"%s\".\"%s\" (%s);\n", mdl.getSchema(), model.getName(), cols));
-            });
-            mdl.listMetrics().forEach(metric -> {
-                String cols = metric.getColumns().stream().map(column -> format("\"%s\" %s", column.getName(), pgMetastore.handlePgType(column.getType()))).collect(joining(","));
-                sb.append(format("CREATE TABLE IF NOT EXISTS \"%s\".\"%s\" (%s);\n", mdl.getSchema(), metric.getName(), cols));
-            });
-            mdl.listCumulativeMetrics().forEach(metric -> {
-                String cols = format("\"%s\" %s, \"%s\" %s",
-                        metric.getMeasure().getName(),
-                        pgMetastore.handlePgType(metric.getMeasure().getType()),
-                        metric.getWindow().getName(),
-                        mdl.getColumnType(metric.getName(), metric.getWindow().getName())
-                                .orElseThrow(() -> new IllegalArgumentException("Unknown window column type")));
-                sb.append(format("CREATE TABLE IF NOT EXISTS \"%s\".\"%s\" (%s);\n", mdl.getSchema(), metric.getName(), cols));
-            });
-            mdl.listViews()
-                    .stream()
-                    .map(view -> {
-                        try {
-                            return Optional.of(new DescribedView(view.getName(), previewService.dryRun(mdl, view.getStatement()).join()));
-                        }
-                        catch (Exception e) {
-                            LOG.error(e, "Failed to describe view %s", view.getName());
-                            return Optional.empty();
-                        }
-                    })
-                    .filter(Optional::isPresent)
-                    .map(Optional::get)
-                    .map(describedView -> (DescribedView) describedView)
-                    .forEach(describedView -> {
-                        String cols = describedView.columns.stream().map(column -> format("\"%s\" %s", column.getName(), column.getType().typName())).collect(joining(","));
-                        sb.append(format("CREATE TABLE IF NOT EXISTS \"%s\".\"%s\" (%s);\n", mdl.getSchema(), describedView.name, cols));
-                    });
-            String syncSql = sb.toString();
-            LOG.info("Sync PG Metastore DDL:\n %s", syncSql);
-            if (!syncSql.isEmpty()) {
-                pgMetastore.directDDL(syncSql);
-            }
-        }
-        catch (Exception e) {
-            // won't throw exception to avoid the sever start failed.
-            LOG.error(e, "Failed to sync PG Metastore");
-            throw new DeployException("Failed to sync PG Metastore", e);
-        }
-    }
-
-    static class DescribedView
-    {
-        private final String name;
-        private final List<Column> columns;
-
-        public DescribedView(String name, List<Column> columns)
-        {
-            this.name = name;
-            this.columns = columns;
-        }
-    }
+    {}
 }