From 7e4c70930f8b35c29013bd20c4930cdb01abc788 Mon Sep 17 00:00:00 2001
From: Ago Allikmaa <ago.allikmaa@eesti.ee>
Date: Wed, 1 Aug 2018 13:06:06 +0300
Subject: [PATCH] Fix missing wildcard from LIKE when querying source segments
 for allocation.

---
 .../IndexerSQLMetadataStorageCoordinator.java         | 11 ++++++++++-
 1 file changed, 10 insertions(+), 1 deletion(-)

diff --git a/server/src/main/java/io/druid/metadata/IndexerSQLMetadataStorageCoordinator.java b/server/src/main/java/io/druid/metadata/IndexerSQLMetadataStorageCoordinator.java
index 99ee8161d8d9..07e1dffd6031 100644
--- a/server/src/main/java/io/druid/metadata/IndexerSQLMetadataStorageCoordinator.java
+++ b/server/src/main/java/io/druid/metadata/IndexerSQLMetadataStorageCoordinator.java
@@ -334,6 +334,8 @@ private List<DataSegment> collectSourceSegments(
           sourceVersion.lhs.getEnd(), sourceVersion.rhs, NoneShardSpec.instance());
     }
 
+    String identifierPrefix = org.apache.commons.lang.StringUtils.getCommonPrefix(identifiers) + "%";
+
     Query<Map<String, Object>> sql = handle
         .createQuery(
             String.format(
@@ -343,7 +345,7 @@ private List<DataSegment> collectSourceSegments(
             )
         )
         .bind(0, dataSource)
-        .bind(1, org.apache.commons.lang.StringUtils.getCommonPrefix(identifiers));
+        .bind(1, identifierPrefix);
 
     index = 0;
 
@@ -352,6 +354,7 @@ private List<DataSegment> collectSourceSegments(
     }
 
     final List<DataSegment> results = new ArrayList<>();
+    int falsePositiveCount = 0;
 
     try (final ResultIterator<byte[]> dbSegments = sql
         .map(ByteArrayMapper.FIRST)
@@ -367,10 +370,16 @@ private List<DataSegment> collectSourceSegments(
 
         if (sourceVersions.contains(new Pair<>(segment.getInterval(), segment.getVersion()))) {
           results.add(segment);
+        } else {
+          falsePositiveCount++;
         }
       }
     }
 
+    log.debug("Collecting corresponding source segments for segment allocation for [%s], result is [%d] segments and " +
+        "[%d] false positives (identifier prefix [%s]).", sourceVersions, results.size(), falsePositiveCount,
+        identifierPrefix);
+
     return results;
   }