@Test public void validateIngestionConfig() { Schema schema = new Schema.SchemaBuilder().setSchemaName(TABLE_NAME).build(); TableConfig tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME).setIngestionConfig(null).build(); TableConfigUtils.validate(tableConfig, schema); tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME) .setIngestionConfig(new IngestionConfig(null, null)).build(); TableConfigUtils.validate(tableConfig, schema); tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME) .setIngestionConfig(new IngestionConfig(new FilterConfig(null), null)).build(); TableConfigUtils.validate(tableConfig, schema); tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME) .setIngestionConfig(new IngestionConfig(new FilterConfig("startsWith(columnX, \"myPrefix\")"), null)).build(); TableConfigUtils.validate(tableConfig, schema); tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME) .setIngestionConfig(new IngestionConfig(new FilterConfig("Groovy({x == 10}, x)"), null)).build(); TableConfigUtils.validate(tableConfig, schema); tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME) .setIngestionConfig(new IngestionConfig(new FilterConfig("Groovy(badExpr)"), null)).build(); try { TableConfigUtils.validate(tableConfig, schema); Assert.fail("Should fail on invalid filter function string"); } catch (IllegalStateException e) { } tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME) .setIngestionConfig(new IngestionConfig(new FilterConfig("fakeFunction(xx)"), null)).build(); try { TableConfigUtils.validate(tableConfig, schema); Assert.fail("Should fail for invalid filter function"); } catch (IllegalStateException e) { } tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME) .setIngestionConfig(new IngestionConfig(null, Collections.emptyList())).build(); TableConfigUtils.validate(tableConfig, schema); tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME).setIngestionConfig( new IngestionConfig(null, Lists.newArrayList(new TransformConfig("myCol", "reverse(anotherCol)")))).build(); try { TableConfigUtils.validate(tableConfig, schema); Assert.fail("Should fail for transformedColumn not present in schema"); } catch (IllegalStateException e) { } schema = new Schema.SchemaBuilder().setSchemaName(TABLE_NAME).addSingleValueDimension("myCol", FieldSpec.DataType.STRING) .build(); tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME).setIngestionConfig( new IngestionConfig(null, Lists.newArrayList(new TransformConfig("myCol", "reverse(anotherCol)")))).build(); TableConfigUtils.validate(tableConfig, schema); schema = new Schema.SchemaBuilder().setSchemaName(TABLE_NAME).addSingleValueDimension("myCol", FieldSpec.DataType.STRING) .addMetric("transformedCol", FieldSpec.DataType.LONG).build(); tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME).setIngestionConfig( new IngestionConfig(null, Lists.newArrayList(new TransformConfig("myCol", "reverse(anotherCol)"), new TransformConfig("transformedCol", "Groovy({x+y}, x, y)")))).build(); TableConfigUtils.validate(tableConfig, schema); tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME).setIngestionConfig( new IngestionConfig(null, Lists.newArrayList(new TransformConfig(null, "reverse(anotherCol)")))).build(); try { TableConfigUtils.validate(tableConfig, schema); Assert.fail("Should fail for null column name in transform config"); } catch (IllegalStateException e) { } tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME) .setIngestionConfig(new IngestionConfig(null, Lists.newArrayList(new TransformConfig("myCol", null)))).build(); try { TableConfigUtils.validate(tableConfig, schema); Assert.fail("Should fail for null transform function in transform config"); } catch (IllegalStateException e) { } tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME).setIngestionConfig( new IngestionConfig(null, Lists.newArrayList(new TransformConfig("myCol", "fakeFunction(col)")))).build(); try { TableConfigUtils.validate(tableConfig, schema); Assert.fail("Should fail for invalid transform function in transform config"); } catch (IllegalStateException e) { } tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME).setIngestionConfig( new IngestionConfig(null, Lists.newArrayList(new TransformConfig("myCol", "Groovy(badExpr)")))).build(); try { TableConfigUtils.validate(tableConfig, schema); Assert.fail("Should fail for invalid transform function in transform config"); } catch (IllegalStateException e) { } tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME).setIngestionConfig( new IngestionConfig(null, Lists.newArrayList(new TransformConfig("myCol", "reverse(myCol)")))).build(); try { TableConfigUtils.validate(tableConfig, schema); Assert.fail("Should fail due to use of myCol as arguments and columnName"); } catch (IllegalStateException e) { } tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME).setIngestionConfig( new IngestionConfig(null, Lists.newArrayList(new TransformConfig("myCol", "Groovy({x + y + myCol}, x, myCol, y)")))).build(); try { TableConfigUtils.validate(tableConfig, schema); Assert.fail("Should fail due to use of myCol as arguments and columnName"); } catch (IllegalStateException e) { } tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME).setIngestionConfig( new IngestionConfig(null, Lists.newArrayList(new TransformConfig("myCol", "reverse(x)"), new TransformConfig("myCol", "lower(y)")))) .build(); try { TableConfigUtils.validate(tableConfig, schema); Assert.fail("Should fail due to duplicate transform config"); } catch (IllegalStateException e) { } tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME).setIngestionConfig( new IngestionConfig(null, Lists.newArrayList(new TransformConfig("a", "reverse(x)"), new TransformConfig("b", "lower(a)")))).build(); try { TableConfigUtils.validate(tableConfig, schema); Assert.fail("Should fail due to using transformed column 'a' as argument for transform function of column 'b'"); } catch (IllegalStateException e) { } }
private static void validateIngestionConfig(@Nullable IngestionConfig ingestionConfig, @Nullable Schema schema) { if (ingestionConfig != null) { FilterConfig filterConfig = ingestionConfig.getFilterConfig(); if (filterConfig != null) { String filterFunction = filterConfig.getFilterFunction(); if (filterFunction != null) { try { FunctionEvaluatorFactory.getExpressionEvaluator(filterFunction); } catch (Exception e) { throw new IllegalStateException("Invalid filter function " + filterFunction, e); } } } List<TransformConfig> transformConfigs = ingestionConfig.getTransformConfigs(); if (transformConfigs != null) { Set<String> transformColumns = new HashSet<>(); Set<String> argumentColumns = new HashSet<>(); for (TransformConfig transformConfig : transformConfigs) { String columnName = transformConfig.getColumnName(); if (schema != null) { Preconditions.checkState(schema.getFieldSpecFor(columnName) != null, "The destination column of the transform function must be present in the schema"); } String transformFunction = transformConfig.getTransformFunction(); if (columnName == null || transformFunction == null) { throw new IllegalStateException( "columnName/transformFunction cannot be null in TransformConfig " + transformConfig); } if (!transformColumns.add(columnName)) { throw new IllegalStateException("Duplicate transform config found for column '" + columnName + "'"); } FunctionEvaluator expressionEvaluator; try { expressionEvaluator = FunctionEvaluatorFactory.getExpressionEvaluator(transformFunction); } catch (Exception e) { throw new IllegalStateException( "Invalid transform function '" + transformFunction + "' for column '" + columnName + "'"); } List<String> arguments = expressionEvaluator.getArguments(); if (arguments.contains(columnName)) { throw new IllegalStateException( "Arguments of a transform function '" + arguments + "' cannot contain the destination column '" + columnName + "'"); } argumentColumns.addAll(arguments); } if (!Collections.disjoint(transformColumns, argumentColumns)) { throw new IllegalStateException( "Derived columns not supported yet. Cannot use a transform column as argument to another transform functions"); } } } }
TableConfigUtils { private static void validateIngestionConfig(@Nullable IngestionConfig ingestionConfig, @Nullable Schema schema) { if (ingestionConfig != null) { FilterConfig filterConfig = ingestionConfig.getFilterConfig(); if (filterConfig != null) { String filterFunction = filterConfig.getFilterFunction(); if (filterFunction != null) { try { FunctionEvaluatorFactory.getExpressionEvaluator(filterFunction); } catch (Exception e) { throw new IllegalStateException("Invalid filter function " + filterFunction, e); } } } List<TransformConfig> transformConfigs = ingestionConfig.getTransformConfigs(); if (transformConfigs != null) { Set<String> transformColumns = new HashSet<>(); Set<String> argumentColumns = new HashSet<>(); for (TransformConfig transformConfig : transformConfigs) { String columnName = transformConfig.getColumnName(); if (schema != null) { Preconditions.checkState(schema.getFieldSpecFor(columnName) != null, "The destination column of the transform function must be present in the schema"); } String transformFunction = transformConfig.getTransformFunction(); if (columnName == null || transformFunction == null) { throw new IllegalStateException( "columnName/transformFunction cannot be null in TransformConfig " + transformConfig); } if (!transformColumns.add(columnName)) { throw new IllegalStateException("Duplicate transform config found for column '" + columnName + "'"); } FunctionEvaluator expressionEvaluator; try { expressionEvaluator = FunctionEvaluatorFactory.getExpressionEvaluator(transformFunction); } catch (Exception e) { throw new IllegalStateException( "Invalid transform function '" + transformFunction + "' for column '" + columnName + "'"); } List<String> arguments = expressionEvaluator.getArguments(); if (arguments.contains(columnName)) { throw new IllegalStateException( "Arguments of a transform function '" + arguments + "' cannot contain the destination column '" + columnName + "'"); } argumentColumns.addAll(arguments); } if (!Collections.disjoint(transformColumns, argumentColumns)) { throw new IllegalStateException( "Derived columns not supported yet. Cannot use a transform column as argument to another transform functions"); } } } } }
TableConfigUtils { private static void validateIngestionConfig(@Nullable IngestionConfig ingestionConfig, @Nullable Schema schema) { if (ingestionConfig != null) { FilterConfig filterConfig = ingestionConfig.getFilterConfig(); if (filterConfig != null) { String filterFunction = filterConfig.getFilterFunction(); if (filterFunction != null) { try { FunctionEvaluatorFactory.getExpressionEvaluator(filterFunction); } catch (Exception e) { throw new IllegalStateException("Invalid filter function " + filterFunction, e); } } } List<TransformConfig> transformConfigs = ingestionConfig.getTransformConfigs(); if (transformConfigs != null) { Set<String> transformColumns = new HashSet<>(); Set<String> argumentColumns = new HashSet<>(); for (TransformConfig transformConfig : transformConfigs) { String columnName = transformConfig.getColumnName(); if (schema != null) { Preconditions.checkState(schema.getFieldSpecFor(columnName) != null, "The destination column of the transform function must be present in the schema"); } String transformFunction = transformConfig.getTransformFunction(); if (columnName == null || transformFunction == null) { throw new IllegalStateException( "columnName/transformFunction cannot be null in TransformConfig " + transformConfig); } if (!transformColumns.add(columnName)) { throw new IllegalStateException("Duplicate transform config found for column '" + columnName + "'"); } FunctionEvaluator expressionEvaluator; try { expressionEvaluator = FunctionEvaluatorFactory.getExpressionEvaluator(transformFunction); } catch (Exception e) { throw new IllegalStateException( "Invalid transform function '" + transformFunction + "' for column '" + columnName + "'"); } List<String> arguments = expressionEvaluator.getArguments(); if (arguments.contains(columnName)) { throw new IllegalStateException( "Arguments of a transform function '" + arguments + "' cannot contain the destination column '" + columnName + "'"); } argumentColumns.addAll(arguments); } if (!Collections.disjoint(transformColumns, argumentColumns)) { throw new IllegalStateException( "Derived columns not supported yet. Cannot use a transform column as argument to another transform functions"); } } } } private  TableConfigUtils(); }
TableConfigUtils { private static void validateIngestionConfig(@Nullable IngestionConfig ingestionConfig, @Nullable Schema schema) { if (ingestionConfig != null) { FilterConfig filterConfig = ingestionConfig.getFilterConfig(); if (filterConfig != null) { String filterFunction = filterConfig.getFilterFunction(); if (filterFunction != null) { try { FunctionEvaluatorFactory.getExpressionEvaluator(filterFunction); } catch (Exception e) { throw new IllegalStateException("Invalid filter function " + filterFunction, e); } } } List<TransformConfig> transformConfigs = ingestionConfig.getTransformConfigs(); if (transformConfigs != null) { Set<String> transformColumns = new HashSet<>(); Set<String> argumentColumns = new HashSet<>(); for (TransformConfig transformConfig : transformConfigs) { String columnName = transformConfig.getColumnName(); if (schema != null) { Preconditions.checkState(schema.getFieldSpecFor(columnName) != null, "The destination column of the transform function must be present in the schema"); } String transformFunction = transformConfig.getTransformFunction(); if (columnName == null || transformFunction == null) { throw new IllegalStateException( "columnName/transformFunction cannot be null in TransformConfig " + transformConfig); } if (!transformColumns.add(columnName)) { throw new IllegalStateException("Duplicate transform config found for column '" + columnName + "'"); } FunctionEvaluator expressionEvaluator; try { expressionEvaluator = FunctionEvaluatorFactory.getExpressionEvaluator(transformFunction); } catch (Exception e) { throw new IllegalStateException( "Invalid transform function '" + transformFunction + "' for column '" + columnName + "'"); } List<String> arguments = expressionEvaluator.getArguments(); if (arguments.contains(columnName)) { throw new IllegalStateException( "Arguments of a transform function '" + arguments + "' cannot contain the destination column '" + columnName + "'"); } argumentColumns.addAll(arguments); } if (!Collections.disjoint(transformColumns, argumentColumns)) { throw new IllegalStateException( "Derived columns not supported yet. Cannot use a transform column as argument to another transform functions"); } } } } private  TableConfigUtils(); static void validate(TableConfig tableConfig, @Nullable Schema schema); static void validateTableName(TableConfig tableConfig); static void validateRetentionConfig(TableConfig tableConfig); }
TableConfigUtils { private static void validateIngestionConfig(@Nullable IngestionConfig ingestionConfig, @Nullable Schema schema) { if (ingestionConfig != null) { FilterConfig filterConfig = ingestionConfig.getFilterConfig(); if (filterConfig != null) { String filterFunction = filterConfig.getFilterFunction(); if (filterFunction != null) { try { FunctionEvaluatorFactory.getExpressionEvaluator(filterFunction); } catch (Exception e) { throw new IllegalStateException("Invalid filter function " + filterFunction, e); } } } List<TransformConfig> transformConfigs = ingestionConfig.getTransformConfigs(); if (transformConfigs != null) { Set<String> transformColumns = new HashSet<>(); Set<String> argumentColumns = new HashSet<>(); for (TransformConfig transformConfig : transformConfigs) { String columnName = transformConfig.getColumnName(); if (schema != null) { Preconditions.checkState(schema.getFieldSpecFor(columnName) != null, "The destination column of the transform function must be present in the schema"); } String transformFunction = transformConfig.getTransformFunction(); if (columnName == null || transformFunction == null) { throw new IllegalStateException( "columnName/transformFunction cannot be null in TransformConfig " + transformConfig); } if (!transformColumns.add(columnName)) { throw new IllegalStateException("Duplicate transform config found for column '" + columnName + "'"); } FunctionEvaluator expressionEvaluator; try { expressionEvaluator = FunctionEvaluatorFactory.getExpressionEvaluator(transformFunction); } catch (Exception e) { throw new IllegalStateException( "Invalid transform function '" + transformFunction + "' for column '" + columnName + "'"); } List<String> arguments = expressionEvaluator.getArguments(); if (arguments.contains(columnName)) { throw new IllegalStateException( "Arguments of a transform function '" + arguments + "' cannot contain the destination column '" + columnName + "'"); } argumentColumns.addAll(arguments); } if (!Collections.disjoint(transformColumns, argumentColumns)) { throw new IllegalStateException( "Derived columns not supported yet. Cannot use a transform column as argument to another transform functions"); } } } } private  TableConfigUtils(); static void validate(TableConfig tableConfig, @Nullable Schema schema); static void validateTableName(TableConfig tableConfig); static void validateRetentionConfig(TableConfig tableConfig);  }