### Question:
JdbcMetadataHandler extends MetadataHandler { @Override public GetTableResponse doGetTable(final BlockAllocator blockAllocator, final GetTableRequest getTableRequest) { try (Connection connection = jdbcConnectionFactory.getConnection(getCredentialProvider())) { Schema partitionSchema = getPartitionSchema(getTableRequest.getCatalogName()); return new GetTableResponse(getTableRequest.getCatalogName(), getTableRequest.getTableName(), getSchema(connection, getTableRequest.getTableName(), partitionSchema), partitionSchema.getFields().stream().map(Field::getName).collect(Collectors.toSet())); } catch (SQLException sqlException) { throw new RuntimeException(sqlException.getErrorCode() + ": " + sqlException.getMessage()); } } protected  JdbcMetadataHandler(); protected  JdbcMetadataHandler(final DatabaseConnectionConfig databaseConnectionConfig, final JdbcConnectionFactory jdbcConnectionFactory); @VisibleForTesting protected  JdbcMetadataHandler(final DatabaseConnectionConfig databaseConnectionConfig, final AWSSecretsManager secretsManager,
            final AmazonAthena athena, final JdbcConnectionFactory jdbcConnectionFactory); @Override ListSchemasResponse doListSchemaNames(final BlockAllocator blockAllocator, final ListSchemasRequest listSchemasRequest); @Override ListTablesResponse doListTables(final BlockAllocator blockAllocator, final ListTablesRequest listTablesRequest); @Override GetTableResponse doGetTable(final BlockAllocator blockAllocator, final GetTableRequest getTableRequest); abstract Schema getPartitionSchema(final String catalogName); @Override abstract void getPartitions(
            final BlockWriter blockWriter,
            final GetTableLayoutRequest request, QueryStatusChecker queryStatusChecker); @Override abstract GetSplitsResponse doGetSplits(BlockAllocator blockAllocator, GetSplitsRequest getSplitsRequest);  }### Answer:
@Test public void doGetTable() throws SQLException { String[] schema = {"DATA_TYPE", "COLUMN_SIZE", "COLUMN_NAME", "DECIMAL_DIGITS", "NUM_PREC_RADIX"}; Object[][] values = {{Types.INTEGER, 12, "testCol1", 0, 0}, {Types.VARCHAR, 25, "testCol2", 0, 0}, {Types.TIMESTAMP, 93, "testCol3", 0, 0}, {Types.TIMESTAMP_WITH_TIMEZONE, 93, "testCol4", 0, 0}}; AtomicInteger rowNumber = new AtomicInteger(-1); ResultSet resultSet = mockResultSet(schema, values, rowNumber); SchemaBuilder expectedSchemaBuilder = SchemaBuilder.newBuilder(); expectedSchemaBuilder.addField(FieldBuilder.newBuilder("testCol1", org.apache.arrow.vector.types.Types.MinorType.INT.getType()).build()); expectedSchemaBuilder.addField(FieldBuilder.newBuilder("testCol2", org.apache.arrow.vector.types.Types.MinorType.VARCHAR.getType()).build()); PARTITION_SCHEMA.getFields().forEach(expectedSchemaBuilder::addField); Schema expected = expectedSchemaBuilder.build(); TableName inputTableName = new TableName("testSchema", "testTable"); Mockito.when(connection.getMetaData().getColumns("testCatalog", inputTableName.getSchemaName(), inputTableName.getTableName(), null)).thenReturn(resultSet); Mockito.when(connection.getCatalog()).thenReturn("testCatalog"); GetTableResponse getTableResponse = this.jdbcMetadataHandler.doGetTable( this.blockAllocator, new GetTableRequest(this.federatedIdentity, "testQueryId", "testCatalog", inputTableName)); Assert.assertEquals(expected, getTableResponse.getSchema()); Assert.assertEquals(inputTableName, getTableResponse.getTableName()); Assert.assertEquals("testCatalog", getTableResponse.getCatalogName()); }
@Test(expected = RuntimeException.class) public void doGetTableNoColumns() { TableName inputTableName = new TableName("testSchema", "testTable"); this.jdbcMetadataHandler.doGetTable(this.blockAllocator, new GetTableRequest(this.federatedIdentity, "testQueryId", "testCatalog", inputTableName)); }