Fail on unknown JDBC type

presto-base-jdbc/src/main/java/io/prestosql/plugin/jdbc/BaseJdbcClient.java

@@ -63,6 +63,7 @@
 import static com.google.common.collect.ImmutableMap.toImmutableMap;
 import static com.google.common.collect.ImmutableSet.toImmutableSet;
 import static com.google.common.collect.Iterables.getOnlyElement;
+import static io.prestosql.plugin.jdbc.BaseJdbcSessionProperties.getUnsupportedTypeHandlingStrategy;
 import static io.prestosql.plugin.jdbc.JdbcErrorCode.JDBC_ERROR;
 import static io.prestosql.plugin.jdbc.StandardColumnMappings.bigintWriteFunction;
 import static io.prestosql.plugin.jdbc.StandardColumnMappings.booleanWriteFunction;
@@ -248,12 +249,22 @@ public List<JdbcColumnHandle> getColumns(ConnectorSession session, JdbcTableHand
                             resultSet.getInt("DECIMAL_DIGITS"),
                             Optional.empty());
                     Optional<ColumnMapping> columnMapping = toPrestoType(session, connection, typeHandle);
-                    // skip unsupported column types
+                    String columnName = resultSet.getString("COLUMN_NAME");
                     if (columnMapping.isPresent()) {
-                        String columnName = resultSet.getString("COLUMN_NAME");
                         boolean nullable = (resultSet.getInt("NULLABLE") != columnNoNulls);
                         columns.add(new JdbcColumnHandle(columnName, typeHandle, columnMapping.get().getType(), nullable));
                     }
+                    else {
+                        UnsupportedTypeHandlingStrategy unsupportedTypeHandlingStrategy = getUnsupportedTypeHandlingStrategy(session);
+                        switch (unsupportedTypeHandlingStrategy) {
+                            case IGNORE:
+                                break;
+                            case FAIL:
+                                throw new PrestoException(JDBC_ERROR, "Unsupported data type for column: " + columnName);
+                            default:
+                                throw new IllegalStateException("Unknown unsupported type handling strategy: " + unsupportedTypeHandlingStrategy);
+                        }
+                    }
                 }
                 if (columns.isEmpty()) {
                     // In rare cases (e.g. PostgreSQL) a table might have no columns.

presto-base-jdbc/src/main/java/io/prestosql/plugin/jdbc/BaseJdbcConfig.java

@@ -14,13 +14,15 @@
 package io.prestosql.plugin.jdbc;
 
 import io.airlift.configuration.Config;
+import io.airlift.configuration.ConfigDescription;
 import io.airlift.configuration.ConfigSecuritySensitive;
 import io.airlift.units.Duration;
 import io.airlift.units.MinDuration;
 
 import javax.annotation.Nullable;
 import javax.validation.constraints.NotNull;
 
+import static io.prestosql.plugin.jdbc.UnsupportedTypeHandlingStrategy.FAIL;
 import static java.util.concurrent.TimeUnit.MINUTES;
 
 public class BaseJdbcConfig
@@ -32,6 +34,7 @@ public class BaseJdbcConfig
     private String passwordCredentialName;
     private boolean caseInsensitiveNameMatching;
     private Duration caseInsensitiveNameMatchingCacheTtl = new Duration(1, MINUTES);
+    private UnsupportedTypeHandlingStrategy unsupportedTypeHandlingStrategy = FAIL;
 
     @NotNull
     public String getConnectionUrl()
@@ -124,4 +127,18 @@ public BaseJdbcConfig setCaseInsensitiveNameMatchingCacheTtl(Duration caseInsens
         this.caseInsensitiveNameMatchingCacheTtl = caseInsensitiveNameMatchingCacheTtl;
         return this;
     }
+
+    @NotNull
+    public UnsupportedTypeHandlingStrategy getUnsupportedTypeHandlingStrategy()
+    {
+        return unsupportedTypeHandlingStrategy;
+    }
+
+    @Config("unsupported-type.handling-strategy")
+    @ConfigDescription("Configures how unsupported column data types should be handled")
+    public BaseJdbcConfig setUnsupportedTypeHandlingStrategy(UnsupportedTypeHandlingStrategy unsupportedTypeHandlingStrategy)
+    {
+        this.unsupportedTypeHandlingStrategy = unsupportedTypeHandlingStrategy;
+        return this;
+    }
 }

presto-base-jdbc/src/main/java/io/prestosql/plugin/jdbc/BaseJdbcSessionProperties.java

@@ -0,0 +1,55 @@
+/*
+ * 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.prestosql.plugin.jdbc;
+
+import com.google.common.collect.ImmutableList;
+import io.prestosql.spi.connector.ConnectorSession;
+import io.prestosql.spi.session.PropertyMetadata;
+
+import javax.inject.Inject;
+
+import java.util.List;
+
+import static io.prestosql.spi.session.PropertyMetadata.enumProperty;
+
+public final class BaseJdbcSessionProperties
+        implements SessionPropertiesProvider
+{
+    public static final String UNSUPPORTED_TYPE_HANDLING_STRATEGY = "unsupported_type_handling_strategy";
+
+    private final List<PropertyMetadata<?>> sessionProperties;
+
+    @Inject
+    public BaseJdbcSessionProperties(BaseJdbcConfig config)
+    {
+        sessionProperties = ImmutableList.of(
+                enumProperty(
+                        UNSUPPORTED_TYPE_HANDLING_STRATEGY,
+                        "Configures how unsupported column data types should be handled",
+                        UnsupportedTypeHandlingStrategy.class,
+                        config.getUnsupportedTypeHandlingStrategy(),
+                        false));
+    }
+
+    @Override
+    public List<PropertyMetadata<?>> getSessionProperties()
+    {
+        return sessionProperties;
+    }
+
+    public static UnsupportedTypeHandlingStrategy getUnsupportedTypeHandlingStrategy(ConnectorSession session)
+    {
+        return session.getProperty(UNSUPPORTED_TYPE_HANDLING_STRATEGY, UnsupportedTypeHandlingStrategy.class);
+    }
+}

presto-base-jdbc/src/main/java/io/prestosql/plugin/jdbc/JdbcConnector.java

@@ -13,6 +13,7 @@
  */
 package io.prestosql.plugin.jdbc;
 
+import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableSet;
 import io.airlift.bootstrap.LifeCycleManager;
 import io.airlift.log.Logger;
@@ -25,10 +26,12 @@
 import io.prestosql.spi.connector.ConnectorSplitManager;
 import io.prestosql.spi.connector.ConnectorTransactionHandle;
 import io.prestosql.spi.procedure.Procedure;
+import io.prestosql.spi.session.PropertyMetadata;
 import io.prestosql.spi.transaction.IsolationLevel;
 
 import javax.inject.Inject;
 
+import java.util.List;
 import java.util.Optional;
 import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
@@ -53,6 +56,7 @@ public class JdbcConnector
     private final JdbcPageSinkProvider jdbcPageSinkProvider;
     private final Optional<ConnectorAccessControl> accessControl;
     private final Set<Procedure> procedures;
+    private final Set<SessionPropertiesProvider> sessionProperties;
 
     private final ConcurrentMap<ConnectorTransactionHandle, JdbcMetadata> transactions = new ConcurrentHashMap<>();
 
@@ -64,7 +68,8 @@ public JdbcConnector(
             JdbcRecordSetProvider jdbcRecordSetProvider,
             JdbcPageSinkProvider jdbcPageSinkProvider,
             Optional<ConnectorAccessControl> accessControl,
-            Set<Procedure> procedures)
+            Set<Procedure> procedures,
+            Set<SessionPropertiesProvider> sessionProperties)
     {
         this.lifeCycleManager = requireNonNull(lifeCycleManager, "lifeCycleManager is null");
         this.jdbcMetadataFactory = requireNonNull(jdbcMetadataFactory, "jdbcMetadataFactory is null");
@@ -73,6 +78,7 @@ public JdbcConnector(
         this.jdbcPageSinkProvider = requireNonNull(jdbcPageSinkProvider, "jdbcPageSinkProvider is null");
         this.accessControl = requireNonNull(accessControl, "accessControl is null");
         this.procedures = ImmutableSet.copyOf(requireNonNull(procedures, "procedures is null"));
+        this.sessionProperties = requireNonNull(sessionProperties, "sessionProperties is null");
     }
 
     @Override
@@ -142,6 +148,14 @@ public Set<Procedure> getProcedures()
         return procedures;
     }
 
+    @Override
+    public List<PropertyMetadata<?>> getSessionProperties()
+    {
+        ImmutableList.Builder builder = ImmutableList.<PropertyMetadata<?>>builder();
+        this.sessionProperties.forEach(sessionPropertiesProvider -> builder.addAll(sessionPropertiesProvider.getSessionProperties()));
+        return builder.build();
+    }
+
     @Override
     public final void shutdown()
     {

presto-base-jdbc/src/main/java/io/prestosql/plugin/jdbc/JdbcModule.java

@@ -21,6 +21,7 @@
 import com.google.inject.Singleton;
 import io.prestosql.plugin.jdbc.jmx.StatisticsAwareConnectionFactory;
 import io.prestosql.plugin.jdbc.jmx.StatisticsAwareJdbcClient;
+import com.google.inject.multibindings.Multibinder;
 import io.prestosql.spi.connector.ConnectorAccessControl;
 import io.prestosql.spi.procedure.Procedure;
 
@@ -45,6 +46,8 @@ public void configure(Binder binder)
     {
         newOptionalBinder(binder, ConnectorAccessControl.class);
         newSetBinder(binder, Procedure.class);
+        Multibinder<SessionPropertiesProvider> sessionProperties = newSetBinder(binder, SessionPropertiesProvider.class);
+        sessionProperties.addBinding().to(BaseJdbcSessionProperties.class).in(Scopes.SINGLETON);
         binder.bind(JdbcMetadataFactory.class).in(Scopes.SINGLETON);
         binder.bind(JdbcSplitManager.class).in(Scopes.SINGLETON);
         binder.bind(JdbcRecordSetProvider.class).in(Scopes.SINGLETON);

presto-base-jdbc/src/main/java/io/prestosql/plugin/jdbc/SessionPropertiesProvider.java

@@ -0,0 +1,23 @@
+/*
+ * 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.prestosql.plugin.jdbc;
+
+import io.prestosql.spi.session.PropertyMetadata;
+
+import java.util.List;
+
+public interface SessionPropertiesProvider
+{
+    List<PropertyMetadata<?>> getSessionProperties();
+}

presto-base-jdbc/src/main/java/io/prestosql/plugin/jdbc/UnsupportedTypeHandlingStrategy.java

@@ -0,0 +1,21 @@
+/*
+ * 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.prestosql.plugin.jdbc;
+
+public enum UnsupportedTypeHandlingStrategy
+{
+    IGNORE,
+    FAIL,
+    /**/;
+}

presto-base-jdbc/src/test/java/io/prestosql/plugin/jdbc/JdbcQueryRunner.java

@@ -46,6 +46,13 @@ public static DistributedQueryRunner createJdbcQueryRunner(TpchTable<?>... table
 
     public static DistributedQueryRunner createJdbcQueryRunner(Iterable<TpchTable<?>> tables)
             throws Exception
+
+    {
+        return createJdbcQueryRunner(tables, TestingH2JdbcModule.createProperties());
+    }
+
+    public static DistributedQueryRunner createJdbcQueryRunner(Iterable<TpchTable<?>> tables, Map<String, String> properties)
+            throws Exception
     {
         DistributedQueryRunner queryRunner = null;
         try {
@@ -54,7 +61,6 @@ public static DistributedQueryRunner createJdbcQueryRunner(Iterable<TpchTable<?>
             queryRunner.installPlugin(new TpchPlugin());
             queryRunner.createCatalog("tpch", "tpch");
 
-            Map<String, String> properties = TestingH2JdbcModule.createProperties();
             createSchema(properties, "tpch");
 
             queryRunner.installPlugin(new JdbcPlugin("base-jdbc", new TestingH2JdbcModule()));

presto-base-jdbc/src/test/java/io/prestosql/plugin/jdbc/TestBaseJdbcConfig.java

@@ -20,6 +20,8 @@
 
 import java.util.Map;
 
+import static io.prestosql.plugin.jdbc.UnsupportedTypeHandlingStrategy.FAIL;
+import static io.prestosql.plugin.jdbc.UnsupportedTypeHandlingStrategy.IGNORE;
 import static java.util.concurrent.TimeUnit.MINUTES;
 import static java.util.concurrent.TimeUnit.SECONDS;
 
@@ -35,7 +37,8 @@ public void testDefaults()
                 .setUserCredentialName(null)
                 .setPasswordCredentialName(null)
                 .setCaseInsensitiveNameMatching(false)
-                .setCaseInsensitiveNameMatchingCacheTtl(new Duration(1, MINUTES)));
+                .setCaseInsensitiveNameMatchingCacheTtl(new Duration(1, MINUTES))
+                .setUnsupportedTypeHandlingStrategy(FAIL));
     }
 
     @Test
@@ -49,6 +52,7 @@ public void testExplicitPropertyMappings()
                 .put("password-credential-name", "bar")
                 .put("case-insensitive-name-matching", "true")
                 .put("case-insensitive-name-matching.cache-ttl", "1s")
+                .put("unsupported-type.handling-strategy", "IGNORE")
                 .build();
 
         BaseJdbcConfig expected = new BaseJdbcConfig()
@@ -58,7 +62,8 @@ public void testExplicitPropertyMappings()
                 .setUserCredentialName("foo")
                 .setPasswordCredentialName("bar")
                 .setCaseInsensitiveNameMatching(true)
-                .setCaseInsensitiveNameMatchingCacheTtl(new Duration(1, SECONDS));
+                .setCaseInsensitiveNameMatchingCacheTtl(new Duration(1, SECONDS))
+                .setUnsupportedTypeHandlingStrategy(IGNORE);
 
         ConfigAssertions.assertFullMapping(properties, expected);
     }

presto-base-jdbc/src/test/java/io/prestosql/plugin/jdbc/TestJdbcDistributedQueries.java

@@ -13,21 +13,67 @@
  */
 package io.prestosql.plugin.jdbc;
 
+import com.google.common.collect.ImmutableList;
 import io.airlift.tpch.TpchTable;
+import io.prestosql.Session;
 import io.prestosql.tests.AbstractTestQueries;
+import io.prestosql.tests.sql.JdbcSqlExecutor;
+import io.prestosql.tests.sql.TestTable;
+import org.testng.annotations.Test;
 
+import java.util.Map;
+import java.util.Optional;
+import java.util.Properties;
+
+import static io.prestosql.plugin.jdbc.BaseJdbcSessionProperties.UNSUPPORTED_TYPE_HANDLING_STRATEGY;
 import static io.prestosql.plugin.jdbc.JdbcQueryRunner.createJdbcQueryRunner;
+import static io.prestosql.plugin.jdbc.UnsupportedTypeHandlingStrategy.IGNORE;
+import static io.prestosql.tests.sql.TestTable.TABLE_NAME_PLACEHOLDER;
+import static java.lang.String.format;
 
 public class TestJdbcDistributedQueries
         extends AbstractTestQueries
 {
+    private final Map<String, String> properties;
+
     public TestJdbcDistributedQueries()
     {
-        super(() -> createJdbcQueryRunner(TpchTable.getTables()));
+        this(TestingH2JdbcModule.createProperties());
+    }
+
+    public TestJdbcDistributedQueries(Map<String, String> properties)
+    {
+        super(() -> createJdbcQueryRunner(ImmutableList.copyOf(TpchTable.getTables()), properties));
+        this.properties = properties;
     }
 
     @Override
     public void testLargeIn()
     {
     }
+
+    @Test
+    public void testFailureOnUnknown()
+    {
+        try (TestTable table = new TestTable(
+                getSqlExecutor(),
+                "tpch.test_failure_on_unknown_type",
+                format("CREATE TABLE %s (i int, x GEOMETRY)", TABLE_NAME_PLACEHOLDER),
+                Optional.of("(1, 'POINT(7 52)')"))) {
+            assertQueryFails("SELECT * FROM " + table.getName(), "Unsupported data type for column: X");
+            assertQuery(onUnsupportedType(IGNORE), "SELECT * FROM " + table.getName(), "VALUES 1");
+        }
+    }
+
+    private Session onUnsupportedType(UnsupportedTypeHandlingStrategy unsupportedTypeHandlingStrategy)
+    {
+        return Session.builder(getSession())
+                .setCatalogSessionProperty("jdbc", UNSUPPORTED_TYPE_HANDLING_STRATEGY, unsupportedTypeHandlingStrategy.name())
+                .build();
+    }
+
+    private JdbcSqlExecutor getSqlExecutor()
+    {
+        return new JdbcSqlExecutor(properties.get("connection-url"), new Properties());
+    }
 }