Change time column name when reading from external sources in MSQ (#14148)

When ingesting from an external source which already contains a column "__time", currently, the value is dropped. Changing the time column name in the external input slice reader resolves this.

extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/input/external/ExternalInputSliceReader.java

@@ -52,6 +52,7 @@ import org.apache.druid.msq.querykit.LazyResourceHolder;
 import org.apache.druid.msq.util.DimensionSchemaUtils;
 import org.apache.druid.segment.RowAdapters;
 import org.apache.druid.segment.RowBasedSegment;
+import org.apache.druid.segment.column.ColumnHolder;
 import org.apache.druid.segment.column.RowSignature;
 import org.apache.druid.segment.incremental.SimpleRowIngestionMeters;
 import org.apache.druid.timeline.SegmentId;
@@ -120,7 +121,7 @@ public class ExternalInputSliceReader implements InputSliceReader
   )
   {
     final InputRowSchema schema = new InputRowSchema(
-        new TimestampSpec("__dummy__", "auto", DateTimes.utc(0)),
+        new TimestampSpec(ColumnHolder.TIME_COLUMN_NAME, "auto", DateTimes.utc(0)),
         new DimensionsSpec(
             signature.getColumnNames().stream().map(
                 column ->

extensions-core/multi-stage-query/src/test/java/org/apache/druid/msq/exec/MSQInsertTest.java

@@ -137,6 +137,43 @@ public class MSQInsertTest extends MSQTestBase
 
   }
 
+  @Test
+  public void testInsertWithExistingTimeColumn() throws IOException
+  {
+    List<Object[]> expectedRows = ImmutableList.of(
+        new Object[] {1678897351000L, "A"},
+        new Object[] {1679588551000L, "B"},
+        new Object[] {1682266951000L, "C"}
+    );
+
+    RowSignature rowSignature = RowSignature.builder()
+                                            .add("__time", ColumnType.LONG)
+                                            .add("flags", ColumnType.STRING)
+                                            .build();
+
+    final File toRead = MSQTestFileUtils.getResourceAsTemporaryFile(temporaryFolder, this,
+                                                                    "/dataset-with-time-column.json"
+    );
+    final String toReadFileNameAsJson = queryFramework().queryJsonMapper().writeValueAsString(toRead.getAbsolutePath());
+
+    testIngestQuery().setSql(" INSERT INTO foo1 SELECT\n"
+                             + "  __time,\n"
+                             + "  flags\n"
+                             + "FROM TABLE(\n"
+                             + "  EXTERN(\n"
+                             + "    '{ \"files\": [" + toReadFileNameAsJson + "],\"type\":\"local\"}',\n"
+                             + "    '{\"type\": \"json\"}',\n"
+                             + "    '[{\"name\": \"__time\", \"type\": \"long\"}, {\"name\": \"flags\", \"type\": \"string\"}]'\n"
+                             + "  )\n"
+                             + ") PARTITIONED BY day")
+                     .setQueryContext(context)
+                     .setExpectedResultRows(expectedRows)
+                     .setExpectedDataSource("foo1")
+                     .setExpectedRowSignature(rowSignature)
+                     .verifyResults();
+
+  }
+
   @Test
   public void testInsertOnExternalDataSource() throws IOException
   {

extensions-core/multi-stage-query/src/test/resources/dataset-with-time-column.json

@@ -0,0 +1,3 @@
+{"__time":"1678897351000","flags":"A"}
+{"__time":"1679588551000","flags":"B"}
+{"__time":"1682266951000","flags":"C"}