public Set<Integer> search(Map<String, String> userProfiles, Map<Assignment, List<Integer>> postingList, int size) {
        List<List<Integer>> includeList = new ArrayList<>();
        Set<Integer> excludeSet = new HashSet<>();
        // 匹配
        for (Assignment assignment : postingList.keySet()) {
            String value = userProfiles.get(assignment.getAttr());
            if (assignment.getValue().equalsIgnoreCase(value)) {
                if ("∈".equals(assignment.getPredicate())) {
                    includeList.add(postingList.get(assignment));
                } else {
                    excludeSet.addAll(postingList.get(assignment));
                }
            }
        }

        // 暴力统计
        // key:conjunctionId, value:频次
        Map<Integer, Integer> conjunctionMap = new HashMap<>();
        for (List<Integer> list : includeList) {
            for (Integer conjunctionId : list) {
                if (excludeSet.contains(conjunctionId)) {
                    continue;
                }
                conjunctionMap.put(conjunctionId, conjunctionMap.getOrDefault(conjunctionId, 0) + 1);
            }
        }
      
        // 过滤
        return conjunctionMap.entrySet().stream()
                .filter(x -> x.getValue() >= size)
                .map(Map.Entry::getKey)
                .collect(Collectors.toSet());
    }

@Slf4j
@Data
public class DNFInvertedIndex implements Serializable {
    // size --> assignmentId  --> conjunctionIdList
    private Map<Integer, Map<Integer, List<Integer>>> index = Maps.newConcurrentMap();

    // conjunctionId --> adIdSet
    private Map<Integer, Set<String>> conjId2AdIdListMap = Maps.newConcurrentMap();
}

    // assignmentId --> assignment
    private Map<Integer, Assignment> assignId2KeyMap = Maps.newConcurrentMap();

    // conjunctionId --> assignmentIdSet
    private Map<Integer, Set<Integer>> conjId2AssignIdMap = Maps.newConcurrentMap();

   // adId --> conjunctionId
    private Map<String, Integer> adId2ConjIdMap = Maps.newConcurrentMap();

    private Cache<String, Boolean> userHitAssignCache = Caffeine.newBuilder()
            .maximumSize(1000000)
            .build();

    // assignmentSet 是广告一个 conjunction 包含的所有定向条件
		public void index(String adId, Set<Assignment> assignmentSet) {
        if (CollectionUtils.isEmpty(assignmentSet)) {
            return;
        }
        Set<Integer> assignIdSet = assignmentSet.stream().map(Assignment::getId).collect(Collectors.toSet());
      	int conjId = getConjunctionId(assignIdSet);
        // 包含定向条件
        Set<Integer> includedAssignIdSet = assignmentSet.stream().filter(Assignment::isIncludeBase).map(Assignment::getId).collect(Collectors.toSet());
        // 排除定向条件
      	Set<Integer> excludedAssignIdSet = assignmentSet.stream().filter(x -> !x.isIncludeBase()).map(Assignment::getId).collect(Collectors.toSet());
				// Z 定向条件处理
        if (includedAssignIdSet.isEmpty()) {
            includedAssignIdSet.add(IndexingFacade.Z_ASSIGN_ID);
            assignIdSet.add(IndexingFacade.Z_ASSIGN_ID);
        }
        addIndex(includedAssignIdSet.size(), includedAssignIdSet, conjId);
        addIndex(0, excludedAssignIdSet, conjId);

        maxSize = Math.max(maxSize, includedAssignIdSet.size());
        adId2ConjIdMap.put(adId, conjId);
        conjId2AdIdListMap.computeIfAbsent(conjId, x -> Sets.newConcurrentHashSet()).add(adId);

        if (!conjId2AssignIdMap.containsKey(conjId)) {
            conjId2AssignIdMap.put(conjId, assignIdSet);
        }
    }

		// 添加索引
    public void addIndex(int size, Set<Integer> assignIdSet, int conjId) {
        if (CollectionUtils.isEmpty(assignIdSet)) {
            return;
        }

        assignIdSet.forEach(assignId -> {
            List<Integer> conjIdList = index
                    .computeIfAbsent(size, x -> Maps.newConcurrentMap())
                    .computeIfAbsent(assignId, y -> Lists.newCopyOnWriteArrayList());
            IndexUtils.addSorted(conjIdList, conjId);
        });
    }

   // 获取或者生成 conjunctionId
    private int getConjunctionId(Set<Integer> assignIdSet) {
        Optional<Integer> existingId = conjId2AssignIdMap.entrySet().stream()
                .filter(entry -> entry.getValue().equals(assignIdSet))
                .map(Map.Entry::getKey)
                .findFirst();
        if (existingId.isPresent()) {
            return existingId.get();
        }

        maxConjunctionId++;
        return maxConjunctionId;
    }

    public List<String> search(Map<String, String> userProfile) {
        Map<String, Set<String>> hitFiledMap = IndexUtils.extractAssignValue(userProfile, assignNameSet);
        int count = hitFiledMap.size();
        Set<Integer> hitConIdSet = new HashSet<>();
        DebugUtils.log(userProfile, "maxSize:{},hitFiledMap:{}", maxSize, hitFiledMap);
        // hit assign
        for (int size = Math.min(maxSize, count); size >= 0; size--) {
            if (size > count || !index.containsKey(size)) {
                continue;
            }
            // assignId  --> conjunctionIdList
            Map<Integer, List<Integer>> assignId2ConIdList = index.get(size);
            if (CollectionUtils.isEmpty(assignId2ConIdList) || assignId2ConIdList.size() < size) {
                continue;
            }

            if (size == 0) {
                hitConIdSet.removeAll(hitConjunctionId(hitFiledMap, assignId2ConIdList, size, userProfile));
                continue;
            }
            hitConIdSet.addAll(hitConjunctionId(hitFiledMap, assignId2ConIdList, size, userProfile));
        }

        // 获取广告
        return hitConIdSet.stream()
                .filter(conjId2AdIdListMap::containsKey)
                .flatMap(conId -> conjId2AdIdListMap.get(conId).stream())
                .collect(Collectors.toList());
    }

    private Set<Integer> hitConjunctionId(Map<String, Set<String>> hitFiledMap, Map<Integer, List<Integer>> assignId2ConIdList, int size, Map<String, String> userProfile) {
        Set<Integer> hitAssignIdSet = hitAssignSet(hitFiledMap, assignId2ConIdList, userProfile);
        DebugUtils.log(userProfile, "size:{},hitAssignIdSet:{}", size, hitAssignIdSet);
        if (hitAssignIdSet.size() < size) {
            return new HashSet<>();
        }

        List<Integer[]> conjunctionIdList = hitAssignIdSet.stream().map(assignId2ConIdList::get).map(x -> x.toArray(new Integer[0])).collect(Collectors.toList());
        // 统计命中的 conjunctionId 个数
        Map<Integer, Integer> conjunctionIdCountMap = IndexUtils.statsCount(conjunctionIdList, size);

        // 过滤小于 size 的 conjunctionId
        Set<Integer> res = conjunctionIdCountMap.keySet().stream().filter(conId -> conjunctionIdCountMap.get(conId) >= size).collect(Collectors.toSet());
        DebugUtils.log(userProfile, "size:{},conjunctionIdCountMap:{},hitConjunctionId:{}", size, conjunctionIdCountMap, res);
        return res;
    }

   public Set<Integer> hitAssignSet(Map<String, Set<String>> hitFiledMap, Map<Integer, List<Integer>> assignId2ConIdList, Map<String, String> userProfile) {
        Set<Integer> res = new HashSet<>(hitFiledMap.size());
        for (Integer assignId : assignId2ConIdList.keySet()) {
            if (assignId == IndexingFacade.Z_ASSIGN_ID) {
                res.add(assignId);
                continue;
            }
            if (!assignId2AssignMap.containsKey(assignId)) {
                continue;
            }
            Assignment assignment = assignId2AssignMap.get(assignId);
            if (!hitFiledMap.containsKey(assignment.getName())) {
                continue;
            }

            Set<String> userTags = hitFiledMap.get(assignment.getName());
            if (CollectionUtils.isEmpty(userTags)) {
                if (userProfileHitAssign(userTags, assignment)) {
                    res.add(assignId);
                }
                continue;
            }

            String cacheKey = getCacheKey(assignment.getName(), userProfile.get(assignment.getName()), assignId);
            Boolean isHit = hitAssignCache(cacheKey);
            if (isHit != null) {
                if (isHit) {
                    res.add(assignId);
                }
                continue;
            }

            isHit = userProfileHitAssign(userTags, assignment);
            userHitAssignCache.put(cacheKey, isHit);

            if (isHit) {
                res.add(assignId);
            }
        }
        return res;
    }


    public Boolean hitAssignCache(String cacheKey) {
        return userHitAssignCache.getIfPresent(cacheKey);
    }

    public String getCacheKey(String key, String value, Integer assignId) {
        return key + "|" + value + "|" + assignId;
    }

    public boolean userProfileHitAssign(Set<String> userFiledData, Assignment assignment) {
        if (!assignment.isIncludeBase() && userFiledData.isEmpty()) {
            return true;
        }

        return assignment.getOperatorEnum().operate(OperatorImpl.builder()
                .userProfileValues(userFiledData)
                .assignValues(assignment.getValues())
                .assignment(assignment)
                .build());
    }

    public boolean update(IndexItem indexItem) {
        String adId = indexItem.getId();
        int conjId = adId2ConjIdMap.get(adId);
        Set<Integer> oldAssignIdSet = conjId2AssignIdMap.get(conjId);
        Set<Integer> newAssignIdSet = indexItem.getAssignmentSet().stream().map(Assignment::getId).collect(Collectors.toSet());
        // 定向条件没有变化
        if (oldAssignIdSet.equals(newAssignIdSet)) {
            return true;
        }

        Set<Integer> newIncludeAssignSet = indexItem.getAssignmentSet().stream().filter(Assignment::isIncludeBase).map(Assignment::getId).collect(Collectors.toSet());
        Set<Integer> newExcludeAssignSet = indexItem.getAssignmentSet().stream().filter(key -> !key.isIncludeBase()).map(Assignment::getId).collect(Collectors.toSet());

        if (newIncludeAssignSet.isEmpty()) {
            newIncludeAssignSet.add(IndexingFacade.Z_ASSIGN_ID);
            newAssignIdSet.add(IndexingFacade.Z_ASSIGN_ID);
        }

        // 当前广告独占 conjId，此时可以复用 conjId
        if (conjId2AdIdListMap.get(conjId).size() == 1 && conjId2AdIdListMap.get(conjId).contains(adId)) {
            log.info("index_update 独占 adId:{},conjId:{},newKeyIds:{},oldKeyIds:{}", indexItem.getId(), conjId, newAssignIdSet, oldAssignIdSet);
            Set<Assignment> oldAssignmentSet = oldAssignIdSet.stream().filter(x -> assignId2AssignMap.containsKey(x)).map(x -> assignId2AssignMap.get(x)).collect(Collectors.toSet());
            Set<Integer> oldIncludeAssignSet = oldAssignmentSet.stream().filter(Assignment::isIncludeBase).map(Assignment::getId).collect(Collectors.toSet());
            Set<Integer> oldExcludeAssignSet = oldAssignmentSet.stream().filter(key -> !key.isIncludeBase()).map(Assignment::getId).collect(Collectors.toSet());

            // 更新 size
            updateAssignSet(oldIncludeAssignSet, newIncludeAssignSet, conjId, oldIncludeAssignSet.size(), newIncludeAssignSet.size());
            updateAssignSet(oldExcludeAssignSet, newExcludeAssignSet, conjId, 0, 0);
            conjId2AssignIdMap.put(conjId, newAssignIdSet);
        } else {
            // 生成新的 newConjId，重新插入，旧的 conjId 其他广告使用，不需要处理
            int newConjId = getConjunctionId(newAssignIdSet);
            log.info("index_update 非独占 adId:{},newConjId:{},oldConjId:{},newKeyIds:{},oldKeyIds:{},otherAdId:{}", indexItem.getId(), newConjId, conjId, newAssignIdSet, oldAssignIdSet, conjId2AdIdListMap.get(conjId));
            addIndex(newIncludeAssignSet.size(), newIncludeAssignSet, newConjId);
            addIndex(0, newExcludeAssignSet, newConjId);

            conjId2AdIdListMap.get(conjId).remove(adId);
            conjId2AdIdListMap.computeIfAbsent(newConjId, x -> Sets.newConcurrentHashSet()).add(adId);
            adId2ConjIdMap.put(adId, newConjId);
            conjId2AssignIdMap.put(newConjId, newAssignIdSet);
        }
        if (newIncludeAssignSet.size() > maxSize) {
            maxSize = getNewMaxSize();
        }
        return true;
    }


    public void updateAssignSet(Set<Integer> oldAssignSet, Set<Integer> newAssignSet, int conjId, int oldSize, int newSize) {
        if (oldSize != newSize) {
            addIndex(newSize, newAssignSet, conjId);
            removeIndex(oldSize, oldAssignSet, conjId);
        } else {
            Pair<Set<Integer>, Set<Integer>> assignDiff = IndexUtils.compareSets(oldAssignSet, newAssignSet);
            Set<Integer> addedAssignIdSet = assignDiff.getLeft();
            Set<Integer> removedAssignIdSet = assignDiff.getRight();
            addIndex(newSize, addedAssignIdSet, conjId);
            removeIndex(newSize, removedAssignIdSet, conjId);
        }
    }

    public void addIndex(int size, Set<Integer> assignIdSet, int conjId) {
        if (CollectionUtils.isEmpty(assignIdSet)) {
            return;
        }

        assignIdSet.forEach(assignId -> {
            List<Integer> conjIdList = index
                    .computeIfAbsent(size, x -> Maps.newConcurrentMap())
                    .computeIfAbsent(assignId, y -> Lists.newCopyOnWriteArrayList());
            IndexUtils.addSorted(conjIdList, conjId);
        });
    }

    public boolean remove(String adId) {
        if (!adId2ConjIdMap.containsKey(adId)) {
            return false;
        }
        int conjId = adId2ConjIdMap.get(adId);

        // 清理 conjunctionId --> adIdList
        if (conjId2AdIdListMap.containsKey(conjId) && conjId2AdIdListMap.get(conjId).contains(adId)) {
            conjId2AdIdListMap.get(conjId).remove(adId);
            if (conjId2AdIdListMap.get(conjId).isEmpty()) {
                conjId2AdIdListMap.remove(conjId);
            }
        }

        // adId --> conjunctionId
        adId2ConjIdMap.remove(adId);

        // conjunctionId 非独占
        if (conjId2AdIdListMap.containsKey(conjId)) {
            log.info("index_delete 非独占 adId:{},conjId:{},keyIds:{},otherAdIds:{}", adId, conjId, conjId2AssignIdMap.get(conjId), conjId2AdIdListMap.get(conjId));
            return true;
        }
        log.info("index_delete 独占 adId:{},conjId:{},keyIds:{},otherAdIds:{}", adId, conjId, conjId2AssignIdMap.get(conjId), conjId2AdIdListMap.get(conjId));
        // 清理 size --> assignId  --> conjunctionIdList
        if (conjId2AssignIdMap.containsKey(conjId)) {
            Set<Integer> assignIdSet = conjId2AssignIdMap.get(conjId);
            Set<Integer> includeAssignIdSet = assignIdSet.stream()
                    .filter(x -> assignId2AssignMap.containsKey(x) && assignId2AssignMap.get(x).isIncludeBase())
                    .collect(Collectors.toSet());
            Set<Integer> excludeAssignIdSet = assignIdSet.stream()
                    .filter(x -> assignId2AssignMap.containsKey(x) && !assignId2AssignMap.get(x).isIncludeBase())
                    .collect(Collectors.toSet());

            removeIndex(includeAssignIdSet.size(), includeAssignIdSet, conjId);
            removeIndex(0, excludeAssignIdSet, conjId);

            // 清理 conjunctionId --> assignIdSet
            conjId2AssignIdMap.remove(conjId);
        }
        maxSize = getNewMaxSize();
        return true;
    }


    public void removeIndex(int size, Set<Integer> assignIdSet, int conjId) {
        if (!index.containsKey(size) || CollectionUtils.isEmpty(assignIdSet)) {
            return;
        }

        Map<Integer, List<Integer>> assignId2ConIdList = index.get(size);
        assignIdSet.stream()
                .filter(assignId2ConIdList::containsKey)
                .peek(assignId -> index.get(size).get(assignId).removeIf(x -> x == conjId))
                .filter(assignId -> index.get(size).get(assignId).isEmpty())
                .forEach(assignId2ConIdList::remove);
        if (assignId2ConIdList.isEmpty()) {
            index.remove(size);
        }
    }

    public static Set<Integer> forceStatsCount(List<Integer[]> list, int size) {
        Map<Integer, Integer> res = new HashMap<>();
        for (Integer[] integers : list) {
            for (Integer integer : integers) {
                res.put(integer, res.getOrDefault(integer, 0) + 1);
            }
        }

        return res.entrySet().stream().filter(e -> e.getValue() >= size).map(Map.Entry::getKey).collect(Collectors.toSet());
    }

    public static Set<Integer> mergeStatsCount(List<Integer[]> list, int size) {
        int n = list.size();  // 数组的数量
        int[] pointers = new int[n];  // 每个数组的指针
        Map<Integer, Integer> countMap = new HashMap<>(100);  // 存储每个数的出现次数

        while (true) {
            int minElement = Integer.MAX_VALUE;

          	// 选取本轮最小值：minElement
            int doneCount = 0;
            for (int i = 0; i < n; i++) {
                if (pointers[i] < list.get(i).length) {
                    minElement = Math.min(minElement, list.get(i)[pointers[i]]);
                } else {
                    doneCount++;
                }
            }

            if (n - doneCount < size || doneCount == n) {
                break;
            }

          	// 统计 minElement 出现次数
            int count = 0;
            for (int i = 0; i < n; i++) {
                if (pointers[i] < list.get(i).length && list.get(i)[pointers[i]] == minElement) {
                    count++;
                    pointers[i]++;
                }
            }

            countMap.put(minElement, countMap.getOrDefault(minElement, 0) + count);
        }

        return countMap.entrySet().stream().filter(e -> e.getValue() >= size).map(Map.Entry::getKey).collect(Collectors.toSet());
    }

    public static Set<Integer> statsCount(List<MutablePair<Integer, Integer[]>> data, int size) {
        Set<Integer> set = new HashSet<>();
        // 排序
        data.sort(Comparator.comparingInt(o -> {
            if (o.getLeft() < o.getRight().length) {
                return o.getRight()[o.getLeft()];
            }
            return Integer.MAX_VALUE;
        }));

        boolean isMatch = false;
        while (data.get(size - 1).getLeft() < data.get(size - 1).getRight().length) {
            Integer nextId = data.get(size - 1).getRight()[data.get(size - 1).getLeft()];
            // conjunctionId 个数是否等于 size
            if (data.get(0).getRight()[data.get(0).getLeft()].equals(nextId)) {
                set.add(data.get(0).getRight()[data.get(0).getLeft()]);
                nextId += 1;
                isMatch = true;
            }

          	// 调整data 中每个数据的：currentId
            skipTo(data, nextId, isMatch, size);
            data.removeIf(x -> x.getLeft() >= x.getRight().length);
            if (data.size() < size) {
                break;
            }

          	// 排序
            data.sort(Comparator.comparingInt(o -> {
                if (o.getLeft() < o.getRight().length) {
                    return o.getRight()[o.getLeft()];
                }
                return Integer.MAX_VALUE;
            }));
        }
        return set;
    }

    public static void skipTo(List<MutablePair<Integer, Integer[]>> data, int targetId, boolean isMatch, int size) {
        // 上次匹配成功，只需要修改前 size 个 MutablePair
        if (isMatch) {
            for (int i = 0; i < size; i++) {
                data.get(i).setLeft(data.get(i).getLeft() + 1);
            }
            return;
        }

        // 修改全部
        for (int i = 0; i < size; i++) {
          	// 找到最近一个比 targetId 大的下标，注意：data.getRight() 是有序的
            int index = IndexUtils.indexedBinarySearch(data.get(i).getRight(), targetId, data.get(i).getLeft(), data.get(i).getRight().length-1);
            // 找到与 targetId 下标相等
            if (index >= 0) {
                data.get(i).setLeft(index);
            } else {
                data.get(i).setLeft(-index - 1);
            }
        }
    }

    public static void main(String[] args) {
        long forceCost = 0;
        long onlineCost = 0;
        long dnfCost = 0;

        for (int i = 0; i < 10000; i++) {
            int size = 30;
            List<Integer[]> mockListData = mockData(10000, size);

            List<MutablePair<Integer, Integer[]>> data = new ArrayList<>();
            for (int j = 0; j < mockListData.size(); j++) {
                data.add(MutablePair.of(0, mockListData.get(j)));
            }

            long start = System.currentTimeMillis();
            Set<Integer> res2 = forceStatsCount(mockListData, size);
            forceCost += System.currentTimeMillis() - start;

            start = System.currentTimeMillis();
            Set<Integer> res3 = mergeStatsCount(mockListData, size);
            onlineCost += System.currentTimeMillis() - start;

            start = System.currentTimeMillis();
            Set<Integer> res1 = dnfStatsCount(data, size);
            dnfCost += System.currentTimeMillis() - start;

            if (!res1.equals(res3) || !res1.equals(res2)) {
                System.out.println("ERROR");
                System.out.println("------------------------------------------");
            }
            if (i % 100 == 0) {
                System.out.println(i + ",forceCost:" + forceCost + ",mergeStatsCount:" + onlineCost + ",dnfCost:" + dnfCost);
            }
        }
        System.out.println("Over");
    }

    public static List<Integer[]> mockData(int count, int size) {
        Map<Integer, List<Integer>> map = new HashMap<>();

        Random random = new Random();
        for (int i = 0; i < count; i++) {
            Set<Integer> set = new HashSet<>();
            while (set.size() < size) {
                set.add(random.nextInt(30));
            }

            for (Integer keyId : set) {
                map.computeIfAbsent(keyId, k -> new ArrayList<>()).add(i);
            }
        }

        List<Integer[]> res = new ArrayList<>();
        for (List<Integer> value : map.values()) {
            res.add(value.stream().sorted(Comparator.comparingInt(o -> o)).toArray(Integer[]::new));
        }

        return res.stream().limit(size + 1).collect(Collectors.toList());
    }