Add keep_expanding_json_column option

README.md

@@ -17,6 +17,7 @@ expand columns having json into multiple columns
   - **name**: name of the column. you can define [JsonPath](http://goessner.net/articles/JsonPath/) style.
   - **type**: type of the column (see below)
   - **format**: format of the timestamp if type is timestamp
+- **keep_expanding_json_column**: Not remove the expanding json column from input schema if it's true (false by default)
 - **stop_on_invalid_record**: Stop bulk load transaction if an invalid record is included (false by default)
 
 ---

src/main/java/org/embulk/filter/expand_json/ExpandJsonFilterPlugin.java

@@ -45,6 +45,10 @@ public interface PluginTask
         @Config("stop_on_invalid_record")
         @ConfigDefault("false")
         boolean getStopOnInvalidRecord();
+
+        @Config("keep_expanding_json_column")
+        @ConfigDefault("false")
+        public boolean getKeepExpandingJsonColumn();
     }
 
     @Override
@@ -80,10 +84,19 @@ private Schema buildOutputSchema(PluginTask task, Schema inputSchema)
         int i = 0; // columns index
         for (Column inputColumn: inputSchema.getColumns()) {
             if (inputColumn.getName().contentEquals(task.getJsonColumnName())) {
-                logger.info("removed column: name: {}, type: {}, index: {}",
+                if (!task.getKeepExpandingJsonColumn()) {
+                    logger.info("removed column: name: {}, type: {}, index: {}",
                             inputColumn.getName(),
                             inputColumn.getType(),
                             inputColumn.getIndex());
+                }
+                else {
+                    logger.info("unchanged expanding column: name: {}, type: {}, index: {}",
+                            inputColumn.getName(),
+                            inputColumn.getType(),
+                            i);
+                    builder.add(new Column(i++, inputColumn.getName(), inputColumn.getType()));
+                }
                 for (ColumnConfig expandedColumnConfig: task.getExpandedColumns()) {
                     logger.info("added column: name: {}, type: {}, options: {}, index: {}",
                                 expandedColumnConfig.getName(),

src/main/java/org/embulk/filter/expand_json/FilteredPageOutput.java

@@ -108,6 +108,7 @@ public Column getOutputColumn()
 
     private final Logger logger = Exec.getLogger(FilteredPageOutput.class);
     private final boolean stopOnInvalidRecord;
+    private final boolean keepExpandingJsonColumn;
     private final List<UnchangedColumn> unchangedColumns;
     private final List<ExpandedColumn> expandedColumns;
     private final Column jsonColumn;
@@ -157,7 +158,7 @@ private List<UnchangedColumn> initializeUnchangedColumns(Schema inputSchema,
         for (Column outputColumn : outputSchema.getColumns()) {
             for (Column inputColumn : inputSchema.getColumns()) {
                 if (inputColumn.getName().equals(outputColumn.getName()) &&
-                        !excludeColumn.getName().equals(outputColumn.getName())) {
+                        (!excludeColumn.getName().equals(outputColumn.getName()) || keepExpandingJsonColumn)) {
 
                     UnchangedColumn unchangedColumn = new UnchangedColumn(outputColumn.getName(),
                                                                           inputColumn,
@@ -191,6 +192,7 @@ private ParseContext initializeParseContext()
     FilteredPageOutput(PluginTask task, Schema inputSchema, Schema outputSchema, PageOutput pageOutput)
     {
         this.stopOnInvalidRecord = task.getStopOnInvalidRecord();
+        this.keepExpandingJsonColumn = task.getKeepExpandingJsonColumn();
         this.jsonColumn = initializeJsonColumn(task, inputSchema);
         this.unchangedColumns = initializeUnchangedColumns(inputSchema,
                                                            outputSchema,

src/test/java/org/embulk/filter/expand_json/TestExpandJsonFilterPlugin.java

@@ -184,6 +184,8 @@ public void testDefaultValue()
         assertEquals("$.", task.getRoot());
         assertEquals("UTC", task.getTimeZone());
         assertEquals("%Y-%m-%d %H:%M:%S.%N %z", task.getDefaultTimestampFormat());
+        assertEquals(false, task.getStopOnInvalidRecord());
+        assertEquals(false, task.getKeepExpandingJsonColumn());
     }
 
     /*
@@ -453,6 +455,60 @@ public void run(TaskSource taskSource, Schema outputSchema)
         });
     }
 
+    @Test
+    public void testExpandedJsonValuesWithKeepJsonColumns()
+    {
+        final String configYaml = "" +
+                "type: expand_json\n" +
+                "json_column_name: _c1\n" +
+                "root: $.\n" +
+                "expanded_columns:\n" +
+                "  - {name: _e0, type: string}\n" +
+                "keep_expanding_json_column: true\n";
+
+        ConfigSource config = getConfigFromYaml(configYaml);
+        final Schema schema = schema("_c0", STRING, "_c1", STRING);
+
+        expandJsonFilterPlugin.transaction(config, schema, new Control()
+        {
+            @Override
+            public void run(TaskSource taskSource, Schema outputSchema)
+            {
+                MockPageOutput mockPageOutput = new MockPageOutput();
+
+                try (PageOutput pageOutput = expandJsonFilterPlugin.open(taskSource, schema, outputSchema, mockPageOutput)) {
+                    for (Page page : PageTestUtils.buildPage(runtime.getBufferAllocator(), schema,
+                            "_v0", "{\"_e0\":\"_ev0\"}")) {
+                        pageOutput.add(page);
+                    }
+
+                    pageOutput.finish();
+                }
+
+                assertEquals(3, outputSchema.getColumnCount());
+                Column column;
+                { // 1st column
+                    column = outputSchema.getColumn(0);
+                    assertTrue(column.getName().equals("_c0") && column.getType().equals(STRING));
+                }
+                { // 2nd column
+                    column = outputSchema.getColumn(1);
+                    assertTrue(column.getName().equals("_c1") && column.getType().equals(STRING));
+                }
+                { // 3rd column
+                    column = outputSchema.getColumn(2);
+                    assertTrue(column.getName().equals("_e0") && column.getType().equals(STRING));
+                }
+
+                for (Object[] record : Pages.toObjects(outputSchema, mockPageOutput.pages)) {
+                    assertEquals("_v0", record[0]);
+                    assertEquals("{\"_e0\":\"_ev0\"}", record[1]);
+                    assertEquals("_ev0", record[2]);
+                }
+            }
+        });
+    }
+
     @Test
     public void testExpandSpecialJsonValuesFromString()
     {