From bc52c57831e431ed3c07ae75fce1e767131a2e45 Mon Sep 17 00:00:00 2001
From: zhengdong <zhengd@outlook.com>
Date: Thu, 5 Jan 2017 15:05:35 +0800
Subject: [PATCH] KYLIN-2356 Incorrect result when filter on numeric columns

---
 .../metadata/filter/FilterCodeSystemFactory.java   | 93 ++++++++++++++++++++++
 .../storage/translate/DerivedFilterTranslator.java |  4 +-
 2 files changed, 95 insertions(+), 2 deletions(-)
 create mode 100755 core-metadata/src/main/java/org/apache/kylin/metadata/filter/FilterCodeSystemFactory.java
 mode change 100644 => 100755 core-storage/src/main/java/org/apache/kylin/storage/translate/DerivedFilterTranslator.java

diff --git a/core-metadata/src/main/java/org/apache/kylin/metadata/filter/FilterCodeSystemFactory.java b/core-metadata/src/main/java/org/apache/kylin/metadata/filter/FilterCodeSystemFactory.java
new file mode 100755
index 0000000..70e57b4
--- /dev/null
+++ b/core-metadata/src/main/java/org/apache/kylin/metadata/filter/FilterCodeSystemFactory.java
@@ -0,0 +1,93 @@
+/*
+ * 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.kylin.metadata.filter;
+
+import java.nio.ByteBuffer;
+import java.util.HashMap;
+
+import org.apache.kylin.common.util.BytesUtil;
+import org.apache.kylin.metadata.datatype.DataType;
+import org.apache.kylin.metadata.model.ColumnDesc;
+
+/**
+ * Created by donald.zheng on 2016/12/19.
+ */
+public class FilterCodeSystemFactory {
+    private final static HashMap<String, IFilterCodeSystem> codeSystemMap = new HashMap<>();
+    static {
+        codeSystemMap.put("string", StringCodeSystem.INSTANCE);
+        codeSystemMap.put("integer", new IFilterCodeSystem() {
+            @Override
+            public boolean isNull(Object code) {
+                return code == null;
+            }
+
+            @Override
+            public void serialize(Object code, ByteBuffer buf) {
+                BytesUtil.writeLong(Long.parseLong(code.toString()), buf);
+            }
+
+            @Override
+            public Object deserialize(ByteBuffer buf) {
+                return BytesUtil.readLong(buf);
+            }
+
+            @Override
+            public int compare(Object o, Object t1) {
+                long l1 = Long.parseLong(o.toString());
+                long l2 = Long.parseLong(t1.toString());
+                return Long.compare(l1, l2);
+            }
+        });
+        codeSystemMap.put("decimal", new IFilterCodeSystem() {
+            @Override
+            public boolean isNull(Object code) {
+                return code == null;
+            }
+
+            @Override
+            public void serialize(Object code, ByteBuffer buf) {
+                BytesUtil.writeUTFString(code.toString(), buf);
+            }
+
+            @Override
+            public Object deserialize(ByteBuffer buf) {
+                return Double.parseDouble(BytesUtil.readUTFString(buf));
+            }
+
+            @Override
+            public int compare(Object o, Object t1) {
+                double d1 = Double.parseDouble(o.toString());
+                double d2 = Double.parseDouble(t1.toString());
+                return Double.compare(d1, d2);
+            }
+        });
+    }
+
+    public static IFilterCodeSystem getFilterCodeSystem(ColumnDesc columnDesc) {
+        DataType dataType = columnDesc.getType();
+        if (dataType.isIntegerFamily()) {
+            return codeSystemMap.get("integer");
+        } else if (dataType.isNumberFamily()) {
+            return codeSystemMap.get("decimal");
+        } else {
+            return codeSystemMap.get("string");
+        }
+    }
+}
diff --git a/core-storage/src/main/java/org/apache/kylin/storage/translate/DerivedFilterTranslator.java b/core-storage/src/main/java/org/apache/kylin/storage/translate/DerivedFilterTranslator.java
old mode 100644
new mode 100755
index 60c3a01..c6b1aac
--- a/core-storage/src/main/java/org/apache/kylin/storage/translate/DerivedFilterTranslator.java
+++ b/core-storage/src/main/java/org/apache/kylin/storage/translate/DerivedFilterTranslator.java
@@ -30,8 +30,8 @@ import org.apache.kylin.dict.lookup.LookupStringTable;
 import org.apache.kylin.metadata.filter.ColumnTupleFilter;
 import org.apache.kylin.metadata.filter.CompareTupleFilter;
 import org.apache.kylin.metadata.filter.ConstantTupleFilter;
+import org.apache.kylin.metadata.filter.FilterCodeSystemFactory;
 import org.apache.kylin.metadata.filter.LogicalTupleFilter;
-import org.apache.kylin.metadata.filter.StringCodeSystem;
 import org.apache.kylin.metadata.filter.TupleFilter;
 import org.apache.kylin.metadata.filter.TupleFilter.FilterOperatorEnum;
 import org.apache.kylin.metadata.model.TblColRef;
@@ -76,7 +76,7 @@ public class DerivedFilterTranslator {
         SingleColumnTuple tuple = new SingleColumnTuple(derivedCol);
         for (String[] row : lookup.getAllRows()) {
             tuple.value = row[di];
-            if (compf.evaluate(tuple, StringCodeSystem.INSTANCE)) {
+            if (compf.evaluate(tuple, FilterCodeSystemFactory.getFilterCodeSystem(derivedCol.getColumnDesc()))) {
                 collect(row, pi, satisfyingHostRecords);
             }
         }
-- 
2.8.3