Merge remote-tracking branch 'apache/master' into windowing-fixes

extensions-core/druid-basic-security/src/main/java/org/apache/druid/security/basic/authentication/BasicHTTPAuthenticator.java

@@ -182,15 +182,27 @@ public void doFilter(ServletRequest servletRequest, ServletResponse servletRespo
         return;
       }
 
-      String[] splits = decodedUserSecret.split(":");
-      if (splits.length != 2) {
+      /* From https://www.rfc-editor.org/rfc/rfc7617.html, we can assume that userid won't include a colon but password
+      can.
+
+       The user-id and password MUST NOT contain any control characters (see
+        "CTL" in Appendix B.1 of [RFC5234]).
+
+       Furthermore, a user-id containing a colon character is invalid, as
+       the first colon in a user-pass string separates user-id and password
+       from one another; text after the first colon is part of the password.
+       User-ids containing colons cannot be encoded in user-pass strings.
+
+       */
+      int split = decodedUserSecret.indexOf(':');
+      if (split < 0) {
         // The decoded user secret is not of the right format
         httpResp.sendError(HttpServletResponse.SC_UNAUTHORIZED);
         return;
       }
 
-      String user = splits[0];
-      char[] password = splits[1].toCharArray();
+      String user = decodedUserSecret.substring(0, split);
+      char[] password = decodedUserSecret.substring(split + 1).toCharArray();
 
       // If any authentication error occurs we send a 401 response immediately and do not proceed further down the filter chain.
       // If the authentication result is null and skipOnFailure property is false, we send a 401 response and do not proceed

extensions-core/druid-basic-security/src/test/java/org/apache/druid/security/authentication/BasicHTTPAuthenticatorTest.java

@@ -112,55 +112,21 @@ public void testGoodPassword() throws IOException, ServletException
   }
 
   @Test
-  public void testGoodPasswordWithValidator() throws IOException, ServletException
+  public void testGoodNonEmptyPasswordWithValidator() throws IOException, ServletException
   {
-    CredentialsValidator validator = EasyMock.createMock(CredentialsValidator.class);
-    BasicHTTPAuthenticator authenticatorWithValidator = new BasicHTTPAuthenticator(
-        CACHE_MANAGER_PROVIDER,
-        "basic",
-        "basic",
-        null,
-        null,
-        false,
-        null, null,
-        false,
-        validator
-    );
-
-    String header = StringUtils.utf8Base64("userA:helloworld");
-    header = StringUtils.format("Basic %s", header);
-
-    EasyMock
-        .expect(
-            validator.validateCredentials(EasyMock.eq("basic"), EasyMock.eq("basic"), EasyMock.eq("userA"), EasyMock.aryEq("helloworld".toCharArray()))
-        )
-        .andReturn(
-            new AuthenticationResult("userA", "basic", "basic", null)
-        )
-        .times(1);
-    EasyMock.replay(validator);
-
-    HttpServletRequest req = EasyMock.createMock(HttpServletRequest.class);
-    EasyMock.expect(req.getHeader("Authorization")).andReturn(header);
-    req.setAttribute(
-        AuthConfig.DRUID_AUTHENTICATION_RESULT,
-        new AuthenticationResult("userA", "basic", "basic", null)
-    );
-    EasyMock.expectLastCall().times(1);
-    EasyMock.replay(req);
-
-    HttpServletResponse resp = EasyMock.createMock(HttpServletResponse.class);
-    EasyMock.replay(resp);
-
-    FilterChain filterChain = EasyMock.createMock(FilterChain.class);
-    filterChain.doFilter(req, resp);
-    EasyMock.expectLastCall().times(1);
-    EasyMock.replay(filterChain);
+    testGoodPasswordWithValidator("userA", "helloworld");
+  }
 
-    Filter authenticatorFilter = authenticatorWithValidator.getFilter();
-    authenticatorFilter.doFilter(req, resp, filterChain);
+  @Test
+  public void testGoodEmptyPasswordWithValidator() throws IOException, ServletException
+  {
+    testGoodPasswordWithValidator("userA", "");
+  }
 
-    EasyMock.verify(req, resp, validator, filterChain);
+  @Test
+  public void testGoodColonInPasswordWithValidator() throws IOException, ServletException
+  {
+    testGoodPasswordWithValidator("userA", "hello:hello");
   }
 
   @Test
@@ -396,4 +362,55 @@ public void testMissingHeader() throws IOException, ServletException
 
     EasyMock.verify(req, resp, filterChain);
   }
+
+  private void testGoodPasswordWithValidator(String username, String password) throws IOException, ServletException
+  {
+    CredentialsValidator validator = EasyMock.createMock(CredentialsValidator.class);
+    BasicHTTPAuthenticator authenticatorWithValidator = new BasicHTTPAuthenticator(
+            CACHE_MANAGER_PROVIDER,
+            "basic",
+            "basic",
+            null,
+            null,
+            false,
+            null, null,
+            false,
+            validator
+    );
+
+    String header = StringUtils.utf8Base64(username + ":" + password);
+    header = StringUtils.format("Basic %s", header);
+
+    EasyMock
+            .expect(
+                    validator.validateCredentials(EasyMock.eq("basic"), EasyMock.eq("basic"), EasyMock.eq(username), EasyMock.aryEq(password.toCharArray()))
+            )
+            .andReturn(
+                    new AuthenticationResult(username, "basic", "basic", null)
+            )
+            .times(1);
+    EasyMock.replay(validator);
+
+    HttpServletRequest req = EasyMock.createMock(HttpServletRequest.class);
+    EasyMock.expect(req.getHeader("Authorization")).andReturn(header);
+    req.setAttribute(
+            AuthConfig.DRUID_AUTHENTICATION_RESULT,
+            new AuthenticationResult(username, "basic", "basic", null)
+    );
+    EasyMock.expectLastCall().times(1);
+    EasyMock.replay(req);
+
+    HttpServletResponse resp = EasyMock.createMock(HttpServletResponse.class);
+    EasyMock.replay(resp);
+
+    FilterChain filterChain = EasyMock.createMock(FilterChain.class);
+    filterChain.doFilter(req, resp);
+    EasyMock.expectLastCall().times(1);
+    EasyMock.replay(filterChain);
+
+    Filter authenticatorFilter = authenticatorWithValidator.getFilter();
+    authenticatorFilter.doFilter(req, resp, filterChain);
+
+    EasyMock.verify(req, resp, validator, filterChain);
+  }
 }

extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/exec/ControllerImpl.java

@@ -167,6 +167,7 @@
 import org.apache.druid.msq.shuffle.input.DurableStorageInputChannelFactory;
 import org.apache.druid.msq.shuffle.input.WorkerInputChannelFactory;
 import org.apache.druid.msq.statistics.PartialKeyStatisticsInformation;
+import org.apache.druid.msq.util.ArrayIngestMode;
 import org.apache.druid.msq.util.DimensionSchemaUtils;
 import org.apache.druid.msq.util.IntervalUtils;
 import org.apache.druid.msq.util.MSQFutureUtils;
@@ -1999,6 +2000,17 @@ private static Pair<List<DimensionSchema>, List<AggregatorFactory>> makeDimensio
       final Query<?> query
   )
   {
+    // Log a warning unconditionally if arrayIngestMode is MVD, since the behaviour is incorrect, and is subject to
+    // deprecation and removal in future
+    if (MultiStageQueryContext.getArrayIngestMode(query.context()) == ArrayIngestMode.MVD) {
+      log.warn(
+          "'%s' is set to 'mvd' in the query's context. This ingests the string arrays as multi-value "
+          + "strings instead of arrays, and is preserved for legacy reasons when MVDs were the only way to ingest string "
+          + "arrays in Druid. It is incorrect behaviour and will likely be removed in the future releases of Druid",
+          MultiStageQueryContext.CTX_ARRAY_INGEST_MODE
+      );
+    }
+
     final List<DimensionSchema> dimensions = new ArrayList<>();
     final List<AggregatorFactory> aggregators = new ArrayList<>();
 
@@ -2076,7 +2088,8 @@ private static Pair<List<DimensionSchema>, List<AggregatorFactory>> makeDimensio
                 DimensionSchemaUtils.createDimensionSchema(
                     outputColumnName,
                     type,
-                    MultiStageQueryContext.useAutoColumnSchemas(query.context())
+                    MultiStageQueryContext.useAutoColumnSchemas(query.context()),
+                    MultiStageQueryContext.getArrayIngestMode(query.context())
                 )
             );
           } else if (!isRollupQuery) {
@@ -2125,7 +2138,8 @@ private static void populateDimensionsAndAggregators(
           DimensionSchemaUtils.createDimensionSchema(
               outputColumn,
               type,
-              MultiStageQueryContext.useAutoColumnSchemas(context)
+              MultiStageQueryContext.useAutoColumnSchemas(context),
+              MultiStageQueryContext.getArrayIngestMode(context)
           )
       );
     }

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

@@ -119,10 +119,9 @@ private static Iterator<SegmentWithDescriptor> inputSourceSegmentIterator(
         new DimensionsSpec(
             signature.getColumnNames().stream().map(
                 column ->
-                    DimensionSchemaUtils.createDimensionSchema(
+                    DimensionSchemaUtils.createDimensionSchemaForExtern(
                         column,
-                        signature.getColumnType(column).orElse(null),
-                        false
+                        signature.getColumnType(column).orElse(null)
                     )
             ).collect(Collectors.toList())
         ),

extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/util/ArrayIngestMode.java

@@ -0,0 +1,45 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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 org.apache.druid.msq.util;
+
+/**
+ * Values that the query context flag 'arrayIngestMode' can take to specify the behaviour of ingestion of arrays via
+ * MSQ's INSERT queries
+ */
+public enum ArrayIngestMode
+{
+  /**
+   * Disables the ingestion of arrays via MSQ's INSERT queries.
+   */
+  NONE,
+
+  /**
+   * String arrays are ingested as MVDs. This is to preserve the legacy behaviour of Druid and will be removed in the
+   * future, since MVDs are not true array types and the behaviour is incorrect.
+   * This also disables the ingestion of numeric arrays
+   */
+  MVD,
+
+  /**
+   * Allows numeric and string arrays to be ingested as arrays. This should be the preferred method of ingestion,
+   * unless bound by compatibility reasons to use 'mvd'
+   */
+  ARRAY
+}

extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/util/DimensionSchemaUtils.java

@@ -24,7 +24,9 @@
 import org.apache.druid.data.input.impl.FloatDimensionSchema;
 import org.apache.druid.data.input.impl.LongDimensionSchema;
 import org.apache.druid.data.input.impl.StringDimensionSchema;
+import org.apache.druid.error.InvalidInput;
 import org.apache.druid.java.util.common.ISE;
+import org.apache.druid.java.util.common.StringUtils;
 import org.apache.druid.segment.AutoTypeColumnSchema;
 import org.apache.druid.segment.DimensionHandlerUtils;
 import org.apache.druid.segment.column.ColumnCapabilities;
@@ -40,15 +42,31 @@
  */
 public class DimensionSchemaUtils
 {
+
+  /**
+   * Creates a dimension schema for creating {@link org.apache.druid.data.input.InputSourceReader}.
+   */
+  public static DimensionSchema createDimensionSchemaForExtern(final String column, @Nullable final ColumnType type)
+  {
+    return createDimensionSchema(
+        column,
+        type,
+        false,
+        // Least restrictive mode since we do not have any type restrictions while reading the extern files.
+        ArrayIngestMode.ARRAY
+    );
+  }
+
   public static DimensionSchema createDimensionSchema(
       final String column,
       @Nullable final ColumnType type,
-      boolean useAutoType
+      boolean useAutoType,
+      ArrayIngestMode arrayIngestMode
   )
   {
     if (useAutoType) {
       // for complex types that are not COMPLEX<json>, we still want to use the handler since 'auto' typing
-      // only works for the 'standard' built-in typesg
+      // only works for the 'standard' built-in types
       if (type != null && type.is(ValueType.COMPLEX) && !ColumnType.NESTED_DATA.equals(type)) {
         final ColumnCapabilities capabilities = ColumnCapabilitiesImpl.createDefault().setType(type);
         return DimensionHandlerUtils.getHandlerFromCapabilities(column, capabilities, null)
@@ -57,35 +75,54 @@ public static DimensionSchema createDimensionSchema(
 
       return new AutoTypeColumnSchema(column);
     } else {
-      // if schema information not available, create a string dimension
+      // if schema information is not available, create a string dimension
       if (type == null) {
         return new StringDimensionSchema(column);
-      }
-
-      switch (type.getType()) {
-        case STRING:
-          return new StringDimensionSchema(column);
-        case LONG:
-          return new LongDimensionSchema(column);
-        case FLOAT:
-          return new FloatDimensionSchema(column);
-        case DOUBLE:
-          return new DoubleDimensionSchema(column);
-        case ARRAY:
-          switch (type.getElementType().getType()) {
-            case STRING:
-              return new StringDimensionSchema(column, DimensionSchema.MultiValueHandling.ARRAY, null);
-            case LONG:
-            case FLOAT:
-            case DOUBLE:
-              return new AutoTypeColumnSchema(column);
-            default:
-              throw new ISE("Cannot create dimension for type [%s]", type.toString());
+      } else if (type.getType() == ValueType.STRING) {
+        return new StringDimensionSchema(column);
+      } else if (type.getType() == ValueType.LONG) {
+        return new LongDimensionSchema(column);
+      } else if (type.getType() == ValueType.FLOAT) {
+        return new FloatDimensionSchema(column);
+      } else if (type.getType() == ValueType.DOUBLE) {
+        return new DoubleDimensionSchema(column);
+      } else if (type.getType() == ValueType.ARRAY) {
+        ValueType elementType = type.getElementType().getType();
+        if (elementType == ValueType.STRING) {
+          if (arrayIngestMode == ArrayIngestMode.NONE) {
+            throw InvalidInput.exception(
+                "String arrays can not be ingested when '%s' is set to '%s'. Either set '%s' in query context "
+                + "to 'array' to ingest the string array as an array, or ingest it as an MVD by explicitly casting the "
+                + "array to an MVD with ARRAY_TO_MV function.",
+                MultiStageQueryContext.CTX_ARRAY_INGEST_MODE,
+                StringUtils.toLowerCase(arrayIngestMode.name()),
+                MultiStageQueryContext.CTX_ARRAY_INGEST_MODE
+            );
+          } else if (arrayIngestMode == ArrayIngestMode.MVD) {
+            return new StringDimensionSchema(column, DimensionSchema.MultiValueHandling.ARRAY, null);
+          } else {
+            // arrayIngestMode == ArrayIngestMode.ARRAY would be true
+            return new AutoTypeColumnSchema(column);
+          }
+        } else if (elementType.isNumeric()) {
+          // ValueType == LONG || ValueType == FLOAT || ValueType == DOUBLE
+          if (arrayIngestMode == ArrayIngestMode.ARRAY) {
+            return new AutoTypeColumnSchema(column);
+          } else {
+            throw InvalidInput.exception(
+                "Numeric arrays can only be ingested when '%s' is set to 'array' in the MSQ query's context. "
+                + "Current value of the parameter [%s]",
+                MultiStageQueryContext.CTX_ARRAY_INGEST_MODE,
+                StringUtils.toLowerCase(arrayIngestMode.name())
+            );
           }
-        default:
-          final ColumnCapabilities capabilities = ColumnCapabilitiesImpl.createDefault().setType(type);
-          return DimensionHandlerUtils.getHandlerFromCapabilities(column, capabilities, null)
-                                      .getDimensionSchema(capabilities);
+        } else {
+          throw new ISE("Cannot create dimension for type [%s]", type.toString());
+        }
+      } else {
+        final ColumnCapabilities capabilities = ColumnCapabilitiesImpl.createDefault().setType(type);
+        return DimensionHandlerUtils.getHandlerFromCapabilities(column, capabilities, null)
+                                    .getDimensionSchema(capabilities);
       }
     }
   }